微服务拓扑

Microservice Topology

微服务拓扑是单个微服务内部的事件驱动拓扑。这定义了要对传入事件执行的数据驱动操作，包括转换、存储和发射。

A microservice topology is the event-driven topology internal to a single microservice. This defines the data-driven operations to be performed on incoming events, including transformation, storage, and emission.

图 2-1显示了从两个输入事件流摄取的单个微服务拓扑。

Figure 2-1 shows a single microservice topology ingesting from two input event streams.

图 2-1。简单的微服务拓扑

微服务拓扑从事件流 A 中摄取事件并将其具体化到数据存储中。本章稍后将更详细地介绍具体化操作。同时，事件流 B 被摄取，某些事件被过滤、转换，然后与存储的状态结合起来。结果输出到新的事件流。微服务的摄取、处理和输出是其拓扑的一部分。

The microservice topology ingests events from event stream A and materializes them into a data store. The materialization operation is covered in greater detail later in this chapter. Meanwhile, event stream B is ingested, and certain events are filtered out, transformed, and then joined against the stored state. The results are output to a new event stream. The ingestion, processing, and output of the microservice are part of its topology.

业务拓扑

Business Topology

业务拓扑是一组实现复杂业务功能的微服务、事件流和 API。它是任意的服务分组，可能代表单个团队或部门拥有的服务，也可能代表实现复杂业务功能超集的服务。第 1 章中详细介绍的业务通信结构构成了业务拓扑。微服务实现业务有界上下文，事件流提供共享跨上下文域数据的数据通信机制。

A business topology is the set of microservices, event streams, and APIs that fulfill complex business functions. It is an arbitrary grouping of services and may represent the services owned by a single team or department or those that fulfill a superset of complex business functionality. The business communication structures detailed in Chapter 1 compose the business topology. Microservices implement the business bounded contexts, and event streams provide the data communication mechanism for sharing cross-context domain data.

图 2-2显示了具有三个独立的微服务和事件流的业务拓扑。请注意，业务拓扑并未详细说明微服务的内部工作原理。

Figure 2-2 shows a business topology with three independent microservices and event streams. Note that the business topology does not detail the inner workings of a microservice.

图 2-2。简单的业务拓扑

微服务 1 消费并转换事件流 A 中的数据，并将结果生成到事件流 B。微服务 2 和微服务 3 都消费事件流 B 中的数据。微服务 2 严格充当消费者，并提供 REST API，可以在其中同步访问数据。同时，微服务3根据其有界上下文要求执行自己的转换，并输出到事件流C。新的微服务和事件流可以根据需要添加到业务拓扑中，通过事件流异步耦合。

Microservice 1 consumes and transforms data from event stream A and produces the results to event stream B. Microservice 2 and microservice 3 both consume from event stream B. Microservice 2 acts strictly as a consumer and provides a REST API in which data can be accessed synchronously. Meanwhile, microservice 3 performs its own transformations according to its bounded context requirements and outputs to event stream C. New microservices and event streams can be added to the business topology as needed, coupled asynchronously through event streams.

未锁定事件

Unkeyed Event

无键事件用于将事件描述为单一的事实陈述。一个示例可以是指示客户与产品交互的事件，例如用户在数字图书平台上打开图书实体。顾名思义，该事件不涉及任何密钥。

Unkeyed events are used to describe an event as a singular statement of fact. An example could be an event indicating that a customer interacted with a product, such as a user opening a book entity on a digital book platform. As the name implies, there is no key involved in this event.

实体事件

Entity Event

实体是唯一的事物，并且以该事物的唯一 ID 为关键。实体事件描述实体（最常见的是业务上下文中的对象）在给定时间点的属性和状态。对于图书出版商来说，一个示例可以是使用 ISBN 键入的图书实体。值字段包含与唯一实体相关的所有必要信息。

An entity is a unique thing and is keyed on the unique ID of that thing. The entity event describes the properties and state of an entity—most commonly an object in the business context—at a given point in time. For a book publisher, an example could be a book entity, keyed on ISBN. The value field contains all the necessary information related to the unique entity.

实体事件在事件驱动架构中尤其重要。它们提供实体状态的连续历史记录，并可用于具体化状态（在下一节中介绍）。只需要最新的实体事件即可确定实体的当前状态。

Entity events are particularly important in event-driven architectures. They provide a continual history of the state of an entity and can be used to materialize state (covered in the next section). Only the latest entity event is needed to determine the current state of an entity.

按键事件

Keyed Event

键控事件包含键，但不代表实体。键控事件通常用于对事件流进行分区，以保证事件流的单个分区内的数据局部性（本章稍后将详细介绍）。一个例子是事件流，以 ISBN 为键，指示哪个用户与该书进行了交互。

A keyed event contains a key but does not represent an entity. Keyed events are usually used for partitioning the stream of events to guarantee data locality within a single partition of an event stream (more on this later in the chapter). An example could be a stream of events, keyed on ISBN, indicating which user has interacted with the book.

请注意，事件可以按键聚合，以便可以为每个 ISBN 组成用户列表，从而产生以ISBN 为键的单个实体事件。

Note that the events could be aggregated by key such that a list of users can be composed for each ISBN, resulting in a single entity event keyed on ISBN..

事件存储和服务

Event Storage and Serving

这些是代理底层数据存储的最低要求：

These are the minimal requirements of the underlying storage of the data by the broker:

分区

Partitioning

事件流可以分为单独的子流，子流的数量可以根据生产者和消费者的需求而变化。这种分区机制允许消费者的多个实例并行处理每个子流，从而实现更大的吞吐量。请注意，队列不需要分区，尽管出于性能目的无论如何对它们进行分区可能很有用。

Event streams can be partitioned into individual substreams, the number of which can vary depending on the needs of the producer and consumer. This partitioning mechanism allows for multiple instances of a consumer to process each substream in parallel, allowing for far greater throughput. Note that queues do not require partitioning, though it may be useful to partition them anyway for performance purposes.

严格订购

Strict ordering

事件流分区中的数据是严格排序的，并且按照与最初发布的顺序完全相同的顺序提供给客户端。

Data in an event stream partition is strictly ordered, and it is served to clients in the exact same order that it was originally published.

不变性

Immutability

所有事件数据一旦发布就完全不可变。没有任何机制可以在事件数据发布后对其进行修改。您只能通过发布包含更新数据的新事件来更改以前的数据。

All event data is completely immutable once published. There is no mechanism that can modify event data once it is published. You can alter previous data only by publishing a new event with the updated data.

索引

Indexing

事件在写入事件流时会被分配一个索引。消费者使用它来管理数据消耗，因为他们可以指定从哪个偏移量开始读取。消费者当前指数与尾部指数之间的差异就是消费者滞后。该指标可用于在消费者数量较高时扩大消费者数量，在消费者数量较低时减少消费者数量。此外，它还可以用于唤醒功能即服务逻辑。

Events are assigned an index when written to the event stream. This is used by the consumers to manage data consumption, as they can specify which offset to begin reading from. The difference between the consumer’s current index and the tail index is the consumer lag. This metric can be used to scale up the number of consumers when it is high, and scale them down when it is low. Additionally, it can also be used to awaken Functions-as-a-Service logic.

无限保留

Infinite retention

事件流必须能够无限期地保留事件。此属性是维护事件流中的状态的基础。

Event streams must be able to retain events for an infinite period of time. This property is foundational for maintaining state in an event stream.

重玩性

Replayability

事件流必须是可重播的，以便任何消费者都可以读取所需的任何数据。这为单一事实来源提供了基础，也是微服务之间通信状态的基础。

Event streams must be replayable, such that any consumer can read whatever data it requires. This provides the basis for the single source of truth and is foundational for communicating state between microservices.

需要考虑的其他因素

Additional Factors to Consider

选择事件代理时还需要考虑许多其他因素。

There are a number of additional factors to consider in the selection of an event broker.

从不可变日志中消费

Consuming from the Immutable Log

虽然不是明确的标准，但常用的事件代理使用仅附加的不可变日志。事件附加在日志末尾，并给出自动递增索引 ID。数据的使用者使用对索引 ID 的引用来访问数据。然后，事件可以作为事件流或队列使用，具体取决于业务需求和事件代理的可用功能。

Though not a definitive standard, commonly available event brokers use an append-only immutable log. Events are appended at the end of the log and given an autoincrementing index ID. Consumers of the data use a reference to the index ID to access data. Events can then be consumed as either an event stream or a queue, depending on the needs of the business and the available functionality of the event broker.

作为事件流使用

Consuming as an event stream

每个使用者负责将其自己的指针更新为事件流中先前读取的索引。该索引称为offset，是从事件流开头开始的当前事件的测量值。偏移量允许多个消费者相互独立地消费和跟踪他们的进度，如图2-6所示。

Each consumer is responsible for updating its own pointers to previously read indices within the event stream. This index, known as the offset, is the measurement of the current event from the beginning of the event stream. Offsets permit multiple consumers to consume and track their progress independently of one another, as shown in Figure 2-6.

图 2-6。消费者组及其每个分区的偏移量

消费者组允许将多个消费者视为同一逻辑实体，并可用于消息消费的水平扩展。新消费者加入消费者组，导致事件流分区分配重新分配。新消费者仅使用其分配的分区中的事件，就像组中先前的旧消费者实例继续仅使用其剩余分配的分区中的事件一样。通过这种方式，可以在同一消费者组之间平衡事件消费，同时确保给定分区的所有事件都由单个消费者实例独占消费。组中活动消费者实例的数量仅限于事件流中分区的数量。

The consumer group allows for multiple consumers to be viewed as the same logical entity and can be leveraged for horizontal scaling of message consumption. A new consumer joins the consumer group, causing a redistribution of event stream partition assignments. The new consumer consumes events only from its assigned partitions, just as older consumer instances previously in the group continue to consume only from their remaining assigned partitions. In this way, event consumption can be balanced across the same consumer group, while ensuring that all events for a given partition are exclusively consumed by a single consumer instance. The number of active consumer instances in the group is limited to the number of partitions in the event stream.

作为队列消费

Consuming as a queue

在基于队列的消费中，每个事件由一个且仅一个微服务实例消费。被消费后，该事件被事件代理标记为“已消费”，并且不再提供给任何其他消费者。作为队列消费时，分区计数并不重要，因为可以使用任意数量的消费者实例进行消费。

In queue-based consumption, each event is consumed by one and only one microservice instance. Upon being consumed, that event is marked as “consumed” by the event broker and is no longer provided to any other consumer. Partition counts do not matter when consuming as a queue, as any number of consumer instances can be used for consumption.

警告

从队列进行处理时，不会维持事件顺序。并行消费者无序地消费和处理事件，而单个消费者可能无法处理某个事件，将其返回到队列以便稍后处理，然后继续处理下一个事件。

Event order is not maintained when processing from a queue. Parallel consumers consume and process events out of order, while a single consumer may fail to process an event, return it to the queue for processing at a later date, and move on to the next event.

并非所有事件代理都支持队列。例如，Apache Pulsar目前支持队列，而阿帕奇卡夫卡没有。图 2-7显示了使用单独偏移确认的队列的实现。

Queues are not supported by all event brokers. For instance, Apache Pulsar currently supports queues while Apache Kafka does not. Figure 2-7 shows the implementation of a queue using individual offset acknowledgment.

图 2-7。将不可变日志作为队列使用

提供单一事实来源

Providing a Single Source of Truth

持久且不可变的日志为单一事实来源提供了存储机制，事件代理成为服务消费和生成数据的唯一位置。这样，就可以保证每个消费者都获得相同的数据副本。

The durable and immutable log provides the storage mechanism for the single source of truth, with the event broker becoming the only location in which services consume and produce data. This way, every consumer is guaranteed to be given an identical copy of the data.

采用事件代理作为单一事实来源需要组织中的文化转变。以前，团队可能只是编写直接 SQL 查询来访问整体数据库中的数据，而现在团队还必须将整体数据发布到事件代理。管理单体应用的开发人员必须确保生成的数据完全准确，因为事件流和单体数据库之间的任何分歧都将被视为生产团队的失败。数据的使用者不再直接耦合在整体上，而是直接从事件流中消费。

Adopting the event broker as the single source of truth requires a culture shift in the organization. Whereas previously a team may simply have written direct SQL queries to access data in a monolith’s database, now the team must also publish the monolith’s data to the event broker. The developers managing the monolith must ensure that the data produced is fully accurate, because any disagreement between the event streams and the monolith’s database will be considered a failure of the producing team. Consumers of the data no longer couple directly on the monolith, but instead consume directly from the event streams.

采用事件驱动的微服务可以创建仅使用事件代理来存储和访问数据的服务。虽然微服务的业务逻辑肯定会使用事件的本地副本，但事件代理仍然是所有数据的单一事实来源。

The adoption of event-driven microservices enables the creation of services that use only the event broker to store and access data. While local copies of the events may certainly be used by the business logic of the microservice, the event broker remains the single source of truth for all data.

将微服务放入容器中

Putting Microservices into Containers

最近由Docker普及的容器将应用程序彼此隔离。容器通过共享内核模型利用现有的主机操作系统。这提供了容器之间的基本分离，而容器本身隔离了环境变量、库和其他依赖项。容器以一小部分成本提供了虚拟机的大部分优势（接下来将介绍），并且启动时间短且资源开销低。

Containers, as recently popularized by Docker, isolate applications from one another. Containers leverage the existing host operating system via a shared kernel model. This provides basic separation between containers, while the container itself isolates environment variables, libraries, and other dependencies. Containers provide most of the benefits of a virtual machine (covered next) at a fraction of the cost, with fast startup times and low resource overhead.

容器的共享操作系统方法确实有一些权衡。容器化应用程序必须能够在主机操作系统上运行。如果应用程序需要专门的操作系统，则需要设置独立的主机。安全性是主要问题之一，因为容器共享对主机操作系统的访问。内核中的漏洞可能会使该主机上的所有容器面临风险。对于友好的工作负载，这不太可能成为问题，但云计算中当前的共享租赁模型开始使其成为一个更大的考虑因素。

Containers’ shared operating system approach does have some tradeoffs. Containerized applications must be able to run on the host OS. If an application requires a specialized OS, then an independent host will need to be set up. Security is one of the major concerns, since containers share access to the host machine’s OS. A vulnerability in the kernel can put all the containers on that host at risk. With friendly workloads this is unlikely to be a problem, but current shared tenancy models in cloud computing are beginning to make it a bigger consideration.

将微服务放入虚拟机中

Putting Microservices into Virtual Machines

虚拟机 (VM) 解决了容器的一些缺点，尽管其采用速度较慢。传统虚拟机通过为每个实例指定的独立操作系统和虚拟化硬件提供完全隔离。尽管这种替代方案提供了比容器更高的安全性，但从历史上看它的成本要高得多。与容器相比，每个虚拟机的开销成本更高，启动时间更慢，系统占用空间更大。

Virtual machines (VMs) address some of the shortcomings of containers, though their adoption has been slower. Traditional VMs provide full isolation with a self-contained OS and virtualized hardware specified for each instance. Although this alternative provides higher security than containers, it has historically been much more expensive. Each VM has higher overhead costs compared to containers, with slower startup times and larger system footprints.

管理容器和虚拟机

Managing Containers and Virtual Machines

容器和虚拟机通过各种称为容器管理系统(CMS) 的专用软件进行管理。它们控制容器部署、资源分配以及与底层计算资源的集成。流行且常用的 CMS 包括 Kubernetes、Docker Engine、Mesos Marathon、Amazon ECS和Nomad。

Containers and VMs are managed through a variety of purpose-built software known as container management systems (CMSes). These control container deployment, resource allocation, and integration with the underlying compute resources. Popular and commonly used CMSes include Kubernetes, Docker Engine, Mesos Marathon, Amazon ECS, and Nomad.

微服务必须能够根据不断变化的工作负载、服务级别协议 (SLA) 和性能要求进行扩展和缩减。必须支持垂直扩展，即在每个微服务实例上增加或减少 CPU、内存和磁盘等计算资源。还必须支持水平扩展，添加或删除新实例。

Microservices must be able to scale up and down depending on changing workloads, service-level agreements (SLAs), and performance requirements. Vertical scaling must be supported, in which compute resources such as CPU, memory, and disk are increased or decreased on each microservice instance. Horizontal scaling must also be supported, with new instances added or removed.

每个微服务应作为单个单元部署。对于许多微服务来说，执行其业务需求只需要一个可执行文件，并且可以将其部署在单个容器中。其他微服务可能更复杂，需要协调多个容器和外部数据存储。这就是 Kubernetes 的 Pod 概念发挥作用的地方，它允许通过单个操作部署和恢复多个容器。Kubernetes 还允许单次运行操作；例如，数据库迁移可以在单个可部署的执行期间运行。

Each microservice should be deployed as a single unit. For many microservices, a single executable is all that is needed to perform its business requirements, and it can be deployed within a single container. Other microservices may be more complex, with multiple containers and external data stores requiring coordination. This is where something like Kubernetes’s pod concept comes into play, allowing for multiple containers to be deployed and reverted as a single action. Kubernetes also allows for single-run operations; for example, database migrations can be run during the execution of the single deployable.

许多实现都支持虚拟机管理，但目前比容器管理受到更多限制。Kubernetes 和 Docker Engine 支持 Google 的 gVisor 和 Kata Containers，而亚马逊的平台则支持 AWS Firecracker。随着开发的继续，容器和虚拟机之间的界限将继续模糊。确保您选择的 CMS 能够处理您需要的容器和虚拟机。

VM management is supported by a number of implementations, but is currently more limited than container management. Kubernetes and Docker Engine support Google’s gVisor and Kata Containers, while Amazon’s platform supports AWS Firecracker. The lines between containers and VMs will continue to blur as development continues. Make sure that the CMS you select will handle the containers and VMs that you require of it.

使用显式模式作为契约

Using Explicit Schemas as Contracts

强制执行数据契约并提供一致性的最佳方法是为每个事件定义一个架构。生产者定义了一个显式模式，详细说明了数据定义和触发逻辑，同一类型的所有事件都遵循此格式。在此过程中，生产者提供了一种将其事件格式传达给所有潜在消费者的机制。反过来，消费者可以自信地根据模式化数据构建他们的微服务业务逻辑。

The best way to enforce data contracts and provide consistency is to define a schema for each event. The producer defines an explicit schema detailing the data definition and the triggering logic, with all events of the same type adhering to this format. In doing so, the producer provides a mechanism for communicating its event format to all prospective consumers. The consumers, in turn, can confidently build their microservice business logic against the schematized data.

消费者必须能够提取其业务流程所需的数据，如果没有对哪些数据应该可用的一系列期望，它就无法做到这一点。消费者通常必须依靠部落知识和团队间沟通来解决数据问题，随着事件流和团队数量的增加，这个过程无法扩展。要求每个消费者独立解释数据也存在很大的风险，因为消费者对数据的解释可能与同行不同，这会导致对单一事实来源的看法不一致。

A consumer must be able to extract the data necessary for its business processes, and it cannot do that without having a set of expectations about what data should be available. Consumers must often rely on tribal knowledge and interteam communication to resolve data issues, a process that is not scalable as the number of event streams and teams increases. There is also substantial risk in requiring each consumer to independently interpret the data, as a consumer may interpret it differently than its peers, which leads to inconsistent views of the single source of truth.

生产者在隐式模式方面也处于劣势。即使有最好的意图，生产者也可能没有注意到（或者他们的单元测试没有揭示）他们已经更改了事件数据定义。如果没有明确检查其服务的事件格式，这种情况可能会被忽视，直到导致下游消费者失败。显式模式为消费者和生产者提供安全性和稳定性。

Producers are also at a disadvantage with implicit schemas. Even with the best of intentions, a producer may not notice (or perhaps their unit tests don’t reveal) that they have altered their event data definition. Without an explicit check of their service’s event format, this situation may go unnoticed until it causes downstream consumers to fail. Explicit schemas give security and stability to both consumers and producers.

模式定义注释

Schema Definition Comments

支持模式定义中的集成注释和任意元数据对于传达事件的含义至关重要。围绕事件的产生和消费的知识应尽可能接近事件定义。模式注释有助于消除数据含义的歧义并减少消费者误解的可能性。有两个主要领域的评论特别有价值：

Support for integrated comments and arbitrary metadata in the schema definition is essential for communicating the meaning of an event. The knowledge surrounding the production and consumption of events should be kept as close as possible to the event definition. Schema comments help remove ambiguity about the data’s meaning and reduce the chance of misinterpretation by consumers. There are two main areas where comments are particularly valuable:

• 指定事件的触​​发逻辑。这通常在模式定义顶部的块头中完成，并且应该清楚地说明生成事件的原因。

• Specifying the triggering logic of the event. This is typically done in a block header at the top of the schema definition and should clearly state why an event has been generated.

• 提供有关结构化模式中特定字段的上下文和清晰度。例如，日期时间字段的注释可以指定格式是 UTC、ISO 还是 Unix 时间。

• Giving context and clarity about a particular field within the structured schema. For example, a datetime field’s comments could specify if the format is UTC, ISO, or Unix time.

全功能模式演化

Full-Featured Schema Evolution

模式格式必须支持全方位的模式演化规则。模式演变使生产者能够更新其服务的输出格式，同时允许消费者继续不间断地消费事件。业务变更可能需要添加新字段、弃用旧字段或扩展字段范围。模式演化框架确保这些更改可以安全地发生，并且生产者和消费者可以彼此独立地更新。

The schema format must support a full range of schema evolution rules. Schema evolution enables producers to update their service’s output format while allowing consumers to continue consuming the events uninterrupted. Business changes may require that new fields be added, old fields be deprecated, or the scope of a field be expanded. A schema evolution framework ensures that these changes can occur safely and that producers and consumers can be updated independently of one another.

如果没有模式演化支持，服务的更新将变得极其昂贵。生产者和消费者被迫密切协调，以前兼容的旧数据可能不再与当前系统兼容。每当生产者更改数据模式时，期望消费者更新其服务是不合理的。事实上，微服务的一个核心价值主张是，除了特殊情况外，它们应该独立于其他服务的发布周期。

Updates to services become prohibitively expensive without schema evolution support. Producers and consumers are forced to coordinate closely, and old, previously compatible data may no longer be compatible with current systems. It is unreasonable to expect consumers to update their services whenever a producer changes the data schema. In fact, a core value proposition of microservices is that they should be independent of the release cycles of other services except in exceptional cases.

一组明确的模式演化规则对于使消费者和生产者能够及时更新其应用程序大有帮助。这些规则称为兼容性类型。

An explicit set of schema evolution rules goes a long way in enabling both consumers and producers to update their applications in their own time. These rules are known as compatibility types.

向前兼容性

Forward compatibility

允许读取使用较新模式生成的数据，就像使用较旧模式生成的数据一样。在事件驱动的架构中，这是一个特别有用的演化需求，因为最常见的系统更改模式始于生产者更新其数据定义并使用较新的模式生成数据。如果使用者需要访问新字段，则只需更新其架构和代码的副本。

Allows for data produced with a newer schema to be read as though it were produced with an older schema. This is a particularly useful evolutionary requirement in an event-driven architecture, as the most common pattern of system change begins with the producer updating its data definition and producing data with the newer schema. The consumer is required only to update its copy of the schema and code should it need access to the new fields.

向后兼容性

Backward compatibility

允许读取使用较旧模式生成的数据，就好像它是使用较新模式生成的一样。这使得数据使用者能够使用较新的模式来读取较旧的数据。在以下几种情况下，此功能特别有用：

• 消费者期待上游团队提供新功能。如果新的模式已经定义，消费者可以在生产者发布之前发布自己的更新。

• 架构编码数据由部署在客户硬件上的产品发送，例如报告用户指标的手机应用程序。可以对新生产者版本的模式格式进行更新，同时保持与以前版本的兼容性。

• 消费者应用程序可能需要重新处理使用旧模式版本生成的事件流中的数据。模式演变确保消费者可以将其转换为熟悉的版本。如果不遵循向后兼容性，消费者将只能读取最新格式的消息。

Allows for data produced with an older schema to be read as though it were produced with a newer schema. This enables a consumer of data to use a newer schema to read older data. There are several scenarios where this is particularly useful:

• The consumer is expecting a new feature to be delivered by the upstream team. If the new schema is already defined, the consumer can release its own update prior to the producer release.

• Schema-encoded data is sent by a product deployed on customer hardware, such as a cell phone application that reports on user metrics. Updates can be made to the schema format for new producer releases, while maintaining the compatibility with previous releases.

• The consumer application may need to reprocess data in the event stream that was produced with an older schema version. Schema evolution ensures that the consumer can translate it to a familiar version. If backward compatibility is not followed, the consumer will only be able to read messages with the latest format.

完全兼容

Full compatibility

向前兼容性和向后兼容性的结合，这是最有力的保证，也是您应该尽可能使用的保证。您始终可以在以后放宽兼容性要求，但收紧它们通常要困难得多。

The union of forward compatibility and backward compatibility, this is the strongest guarantee and the one you should use whenever possible. You can always loosen the compatibility requirements at a later date, but it is often far more difficult to tighten them.

代码生成器支持

Code Generator Support

代码生成器用于将事件模式转换为给定编程语言的类定义或等效结构。生产者使用此类定义来创建和填充新的事件对象。编译器或序列化器（取决于实现）要求生成器尊重数据类型并填充原始模式中指定的所有不可为空字段。然后，生产者创建的对象将转换为其序列化格式并发送到事件代理，如图3-1所示。

A code generator is used to turn an event schema into a class definition or equivalent structure for a given programming language. This class definition is used by the producer to create and populate new event objects. The producer is required by the compiler or serializer (depending on the implementation) to respect data types and populate all non-nullable fields that are specified in the original schema. The objects created by the producer are then converted into their serialized format and sent to the event broker, as shown in Figure 3-1.

图 3-1。使用代码生成器的生产者事件生产工作流程

事件数据的使用者维护自己的模式版本，该版本通常与生产者的版本相同，但可能是较旧或较新的模式，具体取决于模式演化的使用。如果观察到完全兼容性，则服务可以使用任何版本的架构来生成其定义。使用者读取事件并使用其编码的架构版本对其进行反序列化。事件格式要么与消息一起存储，这在规模上可能非常昂贵，要么存储在架构注册表中并按需访问（请参阅“架构注册表”）。一旦反序列化为其原始格式，事件就可以转换为使用者支持的模式版本。此时，演化规则开始发挥作用，默认值应用于缺失的字段，而未使用的字段则完全丢弃。最后，根据模式生成的类将数据转换为对象。此时，消费者的业务逻辑就可以开始运行了。该流程如图3-2所示。

The consumer of the event data maintains its own version of the schema, which is often the same version as the producer’s but could be an older or newer schema, depending on the usage of schema evolution. If full compatibility is being observed, the service can use any version of the schema to generate its definitions. The consumer reads the event and deserializes it using the schema version that it was encoded with. The event format is stored either alongside the message, which can be prohibitively expensive at scale, or in a schema registry and accessed on-demand (see “Schema Registry”). Once deserialized into its original format, the event can be converted to the version of the schema supported by the consumer. Evolution rules come into play at this point, with defaults being applied to missing fields, and unused fields dropped completely. Finally, the data is converted into an object based on the schema-generated class. At this point, the consumer’s business logic may begin its operations. This process is shown in Figure 3-2.

图 3-2。使用代码生成器的消费者事件消费和转换工作流程；请注意，消费者将事件从生产者创建的模式 v2 转换为消费者使用的模式格式 v1

代码生成器支持的最大好处是能够使用您选择的语言根据类定义编写应用程序。如果您使用的是编译语言，代码生成器会提供编译器检查，以确保您不会错误处理事件类型或丢失任何给定的非事件类型。null数据字段。除非遵守架构，否则您的代码将无法编译，因此如果不遵守架构数据定义，您的应用程序将不会被发布。编译语言和非编译语言都受益于有一个类实现来进行编码。当您尝试将错误的类型传递给构造函数或设置器时，现代 IDE 会通知您，而如果您使用通用格式（例如对象键/值映射），您将不会收到任何通知。降低数据处理不当的风险可以使整个生态系统的数据质量更加一致 。

The biggest benefit of code generator support is being able to write your application against a class definition in the language of your choice. If you are using a compiled language, the code generator provides compiler checks to ensure that you aren’t mishandling event types or missing the population of any given non-null data field. Your code will not compile unless it adheres to the schema, and therefore your application will not be shipped without adhering to the schema data definition. Both compiled and noncompiled languages benefit from having a class implementation to code against. A modern IDE will notify you when you’re trying to pass the wrong types into a constructor or setter, whereas you would receive no notification if you’re instead using a generic format such as an object key/value map. Reducing the risk of mishandling data provides for much more consistent data quality across the ecosystem.

重大架构变更

Breaking Schema Changes

有时，架构定义必须以某种方式进行更改，从而导致破坏性的演化变化。发生这种情况的原因有很多，包括不断变化的业务需求改变了原始域的模型、原始域的范围不正确以及定义模式时的人为错误。虽然可以相当容易地更改生产服务以适应新模式，但对下游消费者的影响各不相同，需要考虑在内。

There are times when the schema definition must change in a way that results in a breaking evolutionary change. This can happen for a number of reasons, including evolving business requirements that alter the model of the original domain, improper scoping of the original domain, and human error while defining the schema. While the producing service can be fairly easily changed to accommodate the new schema, the impacts to downstream consumers vary and need to be taken into account.

虽然要求生产者和消费者之间进行密切协调似乎有些严厉，但数据合同的重新协商和领域模型的改变需要每个人的支持。除了重新协商架构之外，您还需要采取一些额外的步骤来适应新架构和从中创建的新事件流。破坏性架构更改往往对于无限期存在的实体影响很大，但对于给定时间段后过期的事件影响较小。

While it may seem heavy-handed to require intense coordination between producers and consumers, the renegotiation of the data contract and the alteration of the domain model require buy-in from everyone. Aside from renegotiating the schema, you need to take some additional steps to accommodate the new schema and new event streams that are created from it. Breaking schema changes tend to be quite impactful for entities that exist indefinitely, but less so for events that expire after a given period of time.

说真话、全部真话、只讲真话

Tell the Truth, the Whole Truth, and Nothing but the Truth

一个好的事件定义不仅仅是一条指示发生某事的消息，而是对该事件期间发生的所有事情的完整描述。用业务术语来说，这是摄取输入数据并应用业务逻辑时生成的结果数据。该输出事件必须被视为单一事实来源，并且必须被记录为不可变的事实以供下游消费者消费。它对实际发生的事情拥有完全的权威，消费者不需要查阅任何其他数据源就知道发生了这样的事件。

A good event definition is not simply a message indicating that something happened, but rather the complete description of everything that happened during that event. In business terms, this is the resultant data that is produced when input data is ingested and the business logic is applied. This output event must be treated as the single source of truth and must be recorded as an immutable fact for consumption by downstream consumers. It is the full and total authority on what actually occurred, and consumers should not need to consult any other source of data to know that such an event took place.

每个流使用单个事件定义

Use a Singular Event Definition per Stream

事件流应包含代表单个逻辑事件的事件。不建议在事件流中混合不同类型的事件，因为这样做可能会混淆事件是什么以及流代表什么的定义。验证正在生成的模式很困难，因为在这种情况下可能会动态添加新模式。尽管在特殊情况下您可能希望忽略此原则，但在架构工作流程中生成和使用的绝大多数事件流都应具有严格的单一定义 。

An event stream should contain events representing a single logical event. It is not advisable to mix different types of events within an event stream, because doing so can muddle the definitions of what the event is and what the stream represents. It is difficult to validate the schemas being produced, as new schemas may be added dynamically in such a scenario. Though there are special circumstances where you may wish to ignore this principle, the vast majority of event streams produced and consumed within your architectural workflow should each have a strict, single definition.

使用最窄的数据类型

Use the Narrowest Data Types

对事件数据使用最窄的类型。这使您可以依靠代码生成器、语言类型检查（如果支持）和序列化单元测试来检查数据的边界。这听起来很简单，但在很多情况下，当您不使用正确的类型时，就会出现歧义。以下是一些容易避免的现实例子：

Use the narrowest types for your event data. This lets you rely on the code generators, language type checking (if supported), and serialization unit tests to check the boundaries of your data. It sounds simple, but there are many cases where ambiguity can creep in when you don’t use the proper types. Here are a few easily avoidable real-world examples:

用于string存储数值

Using string to store a numeric value

这需要消费者解析字符串并将其转换为数值，并且通常会得出 GPS 坐标。这很容易出错并且容易失败，尤其是在null发送值或空字符串时。

This requires the consumer to parse and convert the string to a numeric value and often comes up with GPS coordinates. This is error prone and subject to failures, especially when a null value or an empty string is sent.

用作integer布尔值

Using integer as a boolean

而0和1分别可以用来表示 false 和 true，什么2意思呢？怎么样-1？

While 0 and 1 can be used to denote false and true, respectively, what does 2 mean? How about -1?

用作string枚举

Using string as an enum

这对于生产者来说是有问题的，因为他们必须确保其发布的值与可接受的伪枚举列表相匹配。拼写错误和不正确的值将不可避免地被引入。对此字段感兴趣的消费者需要知道可能值的范围，这将需要与生产团队交谈，除非在模式的注释中指定。无论哪种情况，这都是隐式定义，因为生产者无法防止字符串中值范围的任何更改。整个方法根本就是不好的做法。

This is problematic for producers, as they must ensure that their published values match an accepted pseudo-enum list. Typos and incorrect values will inevitably be introduced. A consumer interested in this field will need to know the range of possible values, and this will require talking to the producing team, unless it’s specified in the comments of the schema. In either case, this is an implicit definition, since the producers are not guarded against any changes to the range of values in the string. This whole approach is simply bad practice.

通常会避免使用枚举，因为生产者担心创建消费者模式中不存在的新枚举标记。但是，消费者有责任考虑它无法识别的枚举令牌，并确定是否应该使用默认值处理它们，或者只是抛出致命异常并停止处理，直到有人能够弄清楚需要做什么。Protobuf 和 A​​vro 都有处理未知枚举令牌的优雅方法，如果您的事件格式选择了其中任何一个，则应使用它们。

Enums are often avoided because producers fear creating a new enum token that isn’t present in the consumer’s schema. However, the consumer has a responsibility to consider enum tokens that it does not recognize, and determine if it should process them using a default value or simply throw a fatal exception and halt processing until someone can work out what needs to be done. Both Protobuf and Avro have elegant ways of handling unknown enum tokens and should be used if either is selected for your event format.

保持活动单一目的

Keep Events Single-Purpose

一种常见的反模式是type向事件定义添加一个字段，其中不同的type值指示事件的特定子功能。这通常是针对“相似但不同”的数据进行的，并且通常是实施者错误地将事件识别为单一目的的结果。尽管这看起来像是一种节省时间的措施或简化了数据访问模式，但使用type参数重载事件很少是一个好主意。

One common anti-pattern is adding a type field to an event definition, where different type values indicate specific subfeatures of the event. This is generally done for data that is “similar yet different” and is often a result of the implementer incorrectly identifying the events as single-purpose. Though it may seem like a time-saving measure or a simplification of a data access pattern, overloading events with type parameters is rarely a good idea.

这种方法存在几个问题。每个type参数值通常具有根本不同的业务含义，即使其技术表示形式与其他参数值几乎相同。这些含义也可能会随着时间的推移而发生变化，并且事件所涵盖的范围也会逐渐扩大。其中一些类型可能需要添加新参数来跟踪特定于类型的信息，而其他类型则需要单独的参数。最终，您可能会遇到这样的情况：几个非常不同的事件都居住在同一个事件模式中，这使得很难推断事件真正代表什么。

There are several problems with this approach. Each type parameter value usually has a fundamentally different business meaning, even if its technical representation is nearly identical to the others. It is also possible for these meanings to change over time and for the scope that an event covers to creep. Some of these types may require the addition of new parameters to track type-specific information, whereas other types require separate parameters. Eventually you could have a situation where there are several very distinct events all inhabiting the same event schema, making it difficult to reason about what the event truly represents.

这种复杂性不仅影响必须维护和填充这些事件的开发人员，而且还影响数据的消费者，他们需要对发布哪些数据以及发布原因有一致的了解。如果数据合同发生变化，他们希望能够将自己与这些变化隔离开来。添加额外的字段类型要求他们仅过滤他们关心的数据。存在消费者无法完全理解类型的各种含义，从而导致错误消费和逻辑错误代码的风险。对于每个消费者，还必须进行额外的处理以丢弃与该消费者不相关的事件。

This complexity affects not only the developers who must maintain and populate these events, but also the data’s consumers, who need to have a consistent understanding about what data is published and why. If the data contract changes, they expect to be able to isolate themselves from those changes. Adding extra field types requires them to filter for only data that they care about. There is a risk that the consumer will fail to fully understand the various meanings of the types, leading to incorrect consumption and logically wrong code. For each consumer, additional processing must also be done to discard events that aren’t relevant to that consumer.

记住数据契约定义的原则。事件应该与单个业务操作相关，而不是与记录大量数据的通用事件相关。如果您似乎需要具有各种type参数的通用事件，这通常表明您的问题空间和有界上下文没有明确定义。

Remember the principles of the data contract definition. Events should be related to a single business action, not a generic event that records large assortments of data. If it seems like you need a generic event with various type parameters, that’s usually a tell-tale sign that your problem space and bounded context is not well defined.

TypeEnum：书籍、电影
ActionEnum：单击

产品参与度{
  产品 ID：长，
  产品类型：TypeEnum，
  动作类型：ActionEnum
}

TypeEnum: Book, Movie
ActionEnum: Click

ProductEngagement {
  productId: Long,
  productType: TypeEnum,
  actionType: ActionEnum
}

产品参与度{
  产品 ID：长，
  产品类型：TypeEnum，
  动作类型：动作枚举，
  //仅适用于type=Movie
  观看预览：{null，布尔值}
}

ProductEngagement {
  productId: Long,
  productType: TypeEnum,
  actionType: ActionEnum,
  //Only applies to type=Movie
  watchedPreview: {null, Boolean}
}

TypeEnum：书籍、电影
ActionEnum：点击、书签

产品参与度{
  产品 ID：长，
  产品类型：TypeEnum，
  动作类型：动作枚举，
  //仅适用于productType=Movie
  观看预览：{null，布尔值}，
  //仅适用于productType=Book,actionType=Bookmark
  页面 ID：{null，Int}
}

TypeEnum: Book, Movie
ActionEnum: Click, Bookmark

ProductEngagement {
  productId: Long,
  productType: TypeEnum,
  actionType: ActionEnum,
  //Only applies to productType=Movie
  watchedPreview: {null, Boolean},
  //Only applies to productType=Book,actionType=Bookmark
  pageId: {null, Int}
}

电影点击 {
  电影ID：长，
  观看预览：布尔值
}

图书点击{
  书号：长
}

书签书签 {
  书号：长，
  页面ID: Int
}

MovieClick {
  movieId: Long,
  watchedPreview: Boolean
}

BookClick {
  bookId: Long
}

BookBookmark {
  bookId: Long,
  pageId: Int
}

尽量减少活动规模

Minimize the Size of Events

当事件规模较小、定义明确且易于处理时，其效果会很好。不过，大型活动可能而且确实会发生。一般来说，这些较大的事件代表了大量的上下文信息。也许它们包含与给定事件相关的许多数据点，并且只是对发生的事情的非常大的测量。

Events work well when they’re small, well defined, and easily processed. Large events can and do happen, though. Generally these larger events represent a lot of contextual information. Perhaps they comprise many data points that are related to the given event, and are simply a very large measurement of something that occurred.

当您考虑举办大型活动的设计时，需要考虑几个因素。确保数据与事件直接相关且相关。额外的数据可能已添加到事件“以防万一”，但它可能对下游消费者没有任何实际用途。如果您发现所有事件数据确实直接相关，请退一步看看您的问题空间。您的微服务需要访问数据吗？您可能想要评估有界上下文以查看服务是否正在执行合理的工作量。也许可以通过将附加功能拆分为自己的服务来缩小服务范围。

There are several considerations when you’re looking at a design that produces a very large event. Make sure that the data is directly related and relevant to the event. Additional data may have been added to an event “just in case,” but it may not be of any real use to the downstream consumers. If you find that all the event data is indeed directly related, take a step back and look at your problem space. Does your microservice require access to the data? You may want to evaluate the bounded context to see if the service is performing a reasonable amount of work. Perhaps the service could be reduced in scope with additional functionality split off into its own service.

不过，这种情况并不总是可以避免的，一些事件处理器会生成非常大的输出文件（可能是一个大图像），这些文件太大而无法放入事件流的单个消息中。在这些场景中，您可以使用指向实际数据的指针，但要谨慎使用。这种方法增加了多个事实来源和有效负载可变性的风险，因为不可变的账本无法确保在其系统之外保存数据。

This scenario is not always avoidable, though—some event processors produce very large output files (perhaps a large image) that are much too big to fit into a single message of an event stream. In these scenarios you can use a pointer to the actual data, but do this sparingly. This approach adds risk in the form of multiple sources of truth and payload mutability, as an immutable ledger cannot ensure the preservation of data outside of its system.

让潜在消费者参与活动设计

Involve Prospective Consumers in the Event Design

在设计新活动时，让该数据的任何预期消费者参与进来非常重要。消费者将比生产者更好地了解自己的需求和预期的业务功能，并可能有助于澄清需求。消费者还将更好地了解他们收到的数据。联合会议或讨论可以解决两个 系统之间数据合同的任何问题。

When designing a new event, it is important to involve any anticipated consumers of this data. Consumers will understand their own needs and anticipated business functions better than the producers and may help in clarifying requirements. Consumers will also get a better understanding of the data coming their way. A joint meeting or discussion can shake out any issues around the data contract between the two systems.

避免将事件作为信号量或信号

Avoid Events as Semaphores or Signals

避免使用事件作为信号量或信号。这些事件只是表明发生了某些事情，而不是结果的唯一事实来源。

Avoid using events as a semaphore or a signal. These events simply indicate that something has occurred without being the single source of truth for the results.

考虑一个非常简单的示例，其中系统输出一个事件，指示任意作业的工作已完成。尽管事件本身表明工作已完成，但工作的实际结果并不包含在事件中。这意味着要正确使用此事件，您必须找到已完成的工作实际所在的位置。一旦一段数据有两个真实来源，就会出现一致性问题。

Consider a very simple example where a system outputs an event indicating that work has been completed for an arbitrary job. Although the event itself indicates the work is done, the actual result of the work is not included in the event. This means that to consume this event properly, you must find where the completed work actually resides. Once there are two sources of truth for a piece of data, consistency problems arise.

数据解放的妥协

Compromises for Data Liberation

数据集及其释放的事件流必须保持完全同步，尽管由于事件传播的延迟，此要求仅限于最终一致性。释放的事件流必须具体化回源表的精确副本，并且此属性广泛用于事件驱动的微服务（如第 7 章所述）。相比之下，遗留系统不会从任何事件流重建其数据集，而是通常具有自己的备份和恢复机制，并且从释放的事件流中绝对不读回任何内容。

A data set and its liberated event stream must be kept fully in sync, although this requirement is limited to eventual consistency due to the latency of event propagation. A stream of liberated events must materialize back into an exact replica of the source table, and this property is used extensively for event-driven microservices (as covered in Chapter 7). In contrast, legacy systems do not rebuild their data sets from any event streams, but instead typically have their own backup and restore mechanisms and read absolutely nothing back from the liberated event stream.

在完美的世界中，所有状态都将从事件流的单一事实源创建、管理、维护和恢复。任何共享状态都应首先发布到事件代理，然后具体化回任何需要具体化状态的服务，包括首先生成数据的服务，如图4-3所示。

In the perfect world, all state would be created, managed, maintained, and restored from the single source of truth of the event streams. Any shared state should be published to the event broker first and materialized back to any services that need to materialize the state, including the service that produced the data in the first place, as shown in Figure 4-3.

图 4-3。在实现之前发布到流媒体

虽然在事件代理中维护状态的理想方案对于新的微服务和重构的遗留应用程序来说是可以实现的，但它不一定对所有应用程序都可用或实用。对于除了与变更数据捕获机制的初始集成之外不太可能被重构或更改的服务尤其如此。遗留系统对于组织来说既极其重要，又难以重构，最严重的违规者被认为是一个大泥球。尽管系统很复杂，但其内部数据仍然需要由其他新系统访问。虽然重构可能绝对是可取的，但在现实中存在许多问题阻止这种情况发生：

While the ideal of maintaining state in the event broker is accessible for new microservices and refactored legacy applications, it is not necessarily available or practical for all applications. This is particularly true for services that are unlikely to ever be refactored or changed beyond initial integration with change-data capture mechanisms. Legacy systems can be both extremely important to the organization and prohibitively difficult to refactor, with the worst offenders being considered a big ball of mud. Despite the complexity of a system, their internal data will still need to be accessed by other new systems. While refactoring may absolutely be desirable, there are a number of issues that prevent this from happening in reality:

有限的开发者支持

Limited developer support

许多遗留系统只有最少的开发人员支持，并且需要省力的解决方案来生成解放的数据。

Many legacy systems have minimal developer support and require low-effort solutions to generate liberated data.

重构费用

Expense of refactoring

将现有的应用程序工作流程重新设计为异步事件驱动和同步 MVC（模型-视图-控制器）Web 应用程序逻辑的混合体可能会非常昂贵，尤其是对于复杂的遗留单体而言。

Reworking the preexisting application workflows into a mix of asynchronous event-driven and synchronous MVC (Model-View-Controller) web application logic may be prohibitively expensive, especially for complex legacy monoliths.

遗留支持风险

Legacy support risk

对遗留系统进行的更改可能会产生意想不到的后果，特别是当系统的职责由于技术债务和与其他系统的未识别的点对点连接而不清楚时。

Changes made to legacy systems may have unintended consequences, especially when the system’s responsibilities are unclear due to technical debt and unidentified point-to-point connections with other systems.

这里有一个妥协的机会。您可以使用数据解放模式从数据存储中提取数据并创建必要的事件流。这是单向事件驱动架构的一种形式，因为遗留系统不会从释放的事件流中读回，如图4-3所示。相反，根本目标是通过严格控制事件数据的发布来保持内部数据集与外部事件流同步。事件流最终将与遗留应用程序的内部数据集保持一致，如图4-4所示。

There is an opportunity for compromise here. You can use data liberation patterns to extract the data out of the data store and create the necessary event streams. This is a form of unidirectional event-driven architecture, as the legacy system will not be reading back from the liberated event stream, as shown in Figure 4-3. Instead, the fundamental goal is to keep the internal data set synchronized with the external event stream through strictly controlled publishing of event data. The event stream will be eventually consistent with the internal data set of the legacy application, as shown in Figure 4-4.

图 4-4。释放和具体化两个服务之间的状态

批量装载

Bulk Loading

批量加载查询并加载数据集中的所有数据。当需要在每个轮询间隔加载整个表时以及在正在进行的增量更新之前，执行批量加载。

Bulk loading queries and loads all of the data from the data set. Bulks loads are performed when the entire table needs to be loaded at each polling interval, as well as prior to ongoing incremental updates.

批量加载可能会很昂贵，因为它需要从数据存储中获取整个数据集。对于较小的数据集，这往往不是问题，但大型数据集，尤其是具有数百万或数十亿条记录的数据集，可能很难获取。对于查询和处理非常大的数据集，我建议您研究特定数据存储的最佳实践，因为这些实践可能会因实施而有很大差异。

Bulk loading can be expensive, as it requires obtaining the entire data set from the data store. For smaller data sets this tends not to be a problem, but large data sets, especially those with millions or billions of records, may be difficult to obtain. For querying and processing very large data sets I recommend you research best practices for your particular data store, since these can vary significantly with implementation.

增量时间戳加载

Incremental Timestamp Loading

通过增量时间戳加载，您可以查询并加载自上一个查询结果的最高时间戳以来的所有数据。此方法使用updated-at数据集中的列或字段来跟踪记录上次修改的时间。在每次增量更新期间，仅updated-at查询时间戳晚于上次处理时间的记录。

With incremental timestamp loading, you query and load all data since the highest timestamp of the previous query’s results. This approach uses an updated-at column or field in the data set that keeps track of the time when the record was last modified. During each incremental update, only records with updated-at timestamps later than the last processed time are queried.

自动递增ID加载

Autoincrementing ID Loading

自增ID加载涉及查询并加载所有大于ID最后值的数据。这需要严格排序的自动增量Integer或Long字段。每次增量更新时，只查询ID大于上次处理的ID的记录。这种方法通常用于查询具有不可变记录的表，例如使用发件箱表时（请参阅“使用更改数据捕获日志解放数据”）。

Autoincrementing ID loading involves querying and loading all data larger than the last value of the ID. This requires a strictly ordered autoincrementing Integer or Long field. During each incremental update, only records with an ID larger than the last processed ID are queried. This approach is often used for querying tables with immutable records, such as when using the outbox tables (see “Liberating Data Using Change-Data Capture Logs”).

自定义查询

Custom Querying

自定义查询仅受客户端查询语言的限制。当客户端仅需要较大数据集中的某个数据子集时，或者在连接和非规范化多个表中的数据以避免内部数据模型过度暴露时，通常会使用此方法。例如，用户可以根据特定字段过滤业务合作伙伴数据，其中每个合作伙伴的数据被发送到其自己的事件流。

A custom query is limited only by the client querying language. This approach is often used when the client requires only a certain subset of data from a larger data set, or when joining and denormalizing data from multiple tables to avoid over-exposure of the internal data model. For instance, a user could filter business partner data according to a specific field, where each partner’s data is sent to its own event stream.

增量更新

Incremental Updating

任何增量更新的第一步是确保数据集记录中提供必要的时间戳或自动增量 ID。必须有一个字段，查询可以使用该字段从尚未处理的记录中筛选出已处理的记录。缺少这些字段的数据集需要添加它们，并且需要配置数据存储以填充必要的updated_at 时间戳或自动增量 ID 字段。如果无法将字段添加到数据集中，则基于查询的模式将无法进行增量更新。

The first step of any incremental update is to ensure that the necessary timestamp or autoincrementing ID is available in the records of your data set. There must be a field that the query can use to filter out records it has already processed from those it has yet to process. Data sets that lack these fields will need to have them added, and the data store will need to be configured to populate the necessary updated_at timestamp or the autoincrementing ID field. If the fields cannot be added to the data set, then incremental updates will not be possible with a query-based pattern.

第二步是确定轮询频率和更新延迟。更高频率的更新为下游数据更新提供更低的延迟，尽管这是以数据存储上更大的总负载为代价的。考虑请求之间的间隔是否足以完成所有数据的加载也很重要。在旧查询仍在加载时开始新查询可能会导致竞争条件，即旧数据会覆盖输出事件流中的新数据。

The second step is to determine the frequency of polling and the latency of the updates. Higher-frequency updates provide lower latency for data updates downstream, though this comes at the expense of a larger total load on the data store. It’s also important to consider whether the interval between requests is sufficient to finish loading all of the data. Beginning a new query while the old one is still loading can lead to race conditions, where older data overwrites newer data in the output event streams.

选择增量更新字段并确定更新频率后，最后一步是在启用增量更新之前执行单个批量加载。在进一步增量更新之前，此批量加载必须查询并生成数据集中的所有现有数据。

Once the incremental update field has been selected and the frequency of updates determined, the final step is to perform a single bulk load before enabling incremental updates. This bulk load must query and produce all of the existing data in the data set prior to further incremental updates.

基于查询的更新的好处

Benefits of Query-Based Updating

基于查询的更新具有许多优点，包括：

Query-based updating has a number of advantages, including:

可定制性

Customizability

可以查询任何数据存储，并且可以使用所有用于查询的客户端选项。

Any data store can be queried, and the entire range of client options for querying is available.

独立投票周期

Independent polling periods

特定查询可以更频繁地执行以满足更严格的 SLA（服务级别协议），而其他更昂贵的查询可以不那么频繁地执行以节省资源。

Specific queries can be executed more frequently to meet tighter SLAs (service-level agreements), while other more expensive queries can be executed less frequently to save resources.

内部数据模型的隔离

Isolation of internal data models

关系数据库可以通过使用底层数据的视图或物化视图来提供与内部数据模型的隔离。此技术可用于隐藏不应在数据存储外部公开的域模型信息。

Relational databases can provide isolation from the internal data model by using views or materialized views of the underlying data. This technique can be used to hide domain model information that should not be exposed outside of the data store.

使用数据存储日志的好处

Benefits of Using Data Store Logs

使用数据存储日志的一些好处包括：

Some benefits of using data store logs include:

删除追踪

Delete tracking

二进制日志包含硬记录删除。这些可以转换为删除事件，而无需像基于查询的更新那样进行软删除。

Binary logs contain hard record deletions. These can be converted into delete events without the need for soft deletes as in query-based updates.

对数据存储性能的影响最小

Minimal effect on data store performance

对于使用预写和二进制日志的数据存储，可以执行更改数据捕获，而不会对数据存储的性能产生任何影响。对于那些使用更改表的系统（例如在 SQL Server 中），影响与数据量有关。

For data stores that use write-ahead and binary logs, change-data capture can be performed without any impact to the data store’s performance. For those that use change tables, such as in SQL Server, the impact is related to the volume of data.

低延迟更新

Low-latency updates

一旦事件写入二进制和预写日志，就可以传播更新。与其他数据释放模式相比，这会导致非常低的延迟。

Updates can be propagated as soon as the event is written to the binary and write-ahead logs. This results in very low latency when compared to other data liberation patterns.

使用数据库日志的缺点

Drawbacks of Using Data Base Logs

以下是使用数据库日志的一些缺点：

The following are some of the downsides to using data base logs:

内部数据模型的公开

Exposure of internal data models

内部数据模型在变更日志中完全公开。与基于查询的更新不同，必须仔细且有选择地管理底层数据模型的隔离，其中视图可用于提供隔离。

The internal data model is completely exposed in the changelogs. Isolation of the underlying data model must be carefully and selectively managed, unlike query-based updating, where views can be used to provide isolation.

数据存储之外的反规范化

Denormalization outside of the data store

变更日志仅包含事件数据。一些 CDC 机制可以从物化视图中提取，但对于许多其他机制，非规范化必须在数据存储之外进行。这可能会导致创建高度规范化的事件流，从而需要下游微服务处理外键连接和反规范化。

Changelogs contain only the event data. Some CDC mechanisms can extract from materialized views, but for many others, denormalization must occur outside of the data store. This may lead to the creation of highly normalized event streams, requiring downstream microservices to handle foreign-key joins and denormalization.

数据集模式和输出事件模式之间的脆弱依赖性

Brittle dependency between data set schema and output event schema

与基于查询的数据解放过程非常相似，基于二进制日志的过程存在于数据存储应用程序之外。有效的数据存储更改（例如更改数据集或重新定义字段类型）可能与事件模式的特定演化规则完全不兼容。

Much like the query-based data liberation process, the binary-log-based process exists outside of the data store application. Valid data store changes, such as altering a data set or redefining a field type, may be completely incompatible for the specific evolution rules of the event schema.

性能考虑因素

Performance Considerations

包含发件箱表会给数据存储及其请求处理应用程序带来额外的负载。对于负载最小的小型数据存储，开销可能完全不会被注意到。或者，对于非常大的数据存储，尤其是那些具有大量负载和捕获的许多表的数据存储，它可能非常昂贵。应根据具体情况评估此方法的成本，并与响应策略（例如解析变更数据捕获日志）的成本进行平衡。

The inclusion of outbox tables introduces additional load on the data store and its request-handling applications. For small data stores with minimal load, the overhead may go completely unnoticed. Alternately, it may be quite expensive with very large data stores, particularly those with significant load and many tables under capture. The cost of this approach should be evaluated on a case-by-case basis and balanced against the costs of a reactive strategy such as parsing the change-data capture logs.

隔离内部数据模型

Isolating Internal Data Models

发件箱不需要与内部表进行 1:1 映射。事实上，发件箱的主要好处之一是数据存储客户端可以将内部数据模型与下游消费者隔离。域的内部数据模型可能使用许多高度规范化的表，这些表针对关系操作进行了优化，但很大程度上不适合下游消费者的消费。即使简单的域也可能包含多个表，如果将其作为独立的流公开，则需要重建以供下游消费者使用。就操作开销而言，这很快就会变得极其昂贵，因为多个下游团队将不得不重建域模型并处理事件流中的关系数据。

An outbox does not need to map 1:1 with an internal table. In fact, one of the major benefits of the outbox is that the data store client can isolate the internal data model from downstream consumers. The internal data model of the domain may use a number of highly normalized tables that are optimized for relational operations but are largely unsuitable for consumption by downstream consumers. Even simple domains may comprise multiple tables, which if exposed as independent streams, would require reconstruction for usage by downstream consumers. This quickly becomes extremely expensive in terms of operational overhead, as multiple downstream teams will have to reconstruct the domain model and deal with handling relational data in event streams.

数据存储客户端可以在插入时对数据进行反规范化，以便发件箱反映预期的公共数据契约，尽管这确实以额外的性能和存储空间为代价。另一种选择是维护输出事件流更改的 1:1 映射，并使用专用于此任务的下游事件处理器对流进行非规范化。我将这个过程称为“事件化”，因为它将高度规范化的关系数据转换为易于使用的单个事件更新。这模仿了数据存储客户端可以执行的操作，但在数据存储外部执行此操作以减少负载。图 4-7显示了一个示例，其中根据用户、位置和雇主对用户进行非规范化。

The data store client can instead denormalize data upon insertion time such that the outbox mirrors the intended public data contract, though this does come at the expense of additional performance and storage space. Another option is to maintain the 1:1 mapping of changes to output event streams and denormalize the streams with a downstream event processor dedicated to just this task. This is a process that I call eventification, as it converts highly normalized relational data into easy-to-consume single event updates. This mimics what the data store client could do but does it externally to the data store to reduce load. An example of this is shown in Figure 4-7, where a User is denormalized based on User, Location, and Employer.

图 4-7。使用私有用户、位置和雇主事件流对公共用户事件进行事件化

在此示例中，用户具有对其所居住的城市、州/省和国家/地区的外键引用，以及对其当前雇主的外键引用。用户事件的下游使用者可能只是希望在单个事件中获得有关每个用户的所有信息，而不是被迫将每个流具体化到状态存储中并使用关系工具对其进行非规范化，这是合理的。原始的标准化事件从发件箱发送到它们自己的事件流中，但这些流保存在组织其他部分的私有命名空间中（在“事件流元数据标记”中介绍），以保护内部数据模型。

In this example, the User has a foreign-key reference to the city, state/province, and country they live in, as well as a foreign-key reference to their current employer. It is reasonable that a downstream consumer of a User event may simply want everything about each user in a single event, instead of being forced to materialize each stream into a state store and use relational tooling to denormalize it. The raw, normalized events are sourced from the outboxes into their own event streams, but these streams are kept in a private namespace from the rest of the organization (covered in “Event Stream Metadata Tagging”) to protect the internal data model.

用户的事件化是通过对用户实体进行非规范化并摆脱任何内部数据模型结构来执行的。此过程需要维护用户、位置和雇主的物化表，以便任何更新都可以重新执行连接逻辑并为所有受影响的用户发出更新。最终事件被发送到组织的公共命名空间，供任何下游消费者使用。

Eventification of the user is performed by denormalizing the User entity and shedding any internal data model structures. This process requires maintaining materialized tables of User, Location, and Employer, such that any updates can re-exercise the join logic and emit updates for all affected Users. The final event is emitted to the public namespace of the organization for any downstream consumer to consume.

内部数据模型与外部消费者的隔离程度往往成为转向事件驱动微服务的组织的争论点。隔离内部数据模型对于确保服务的解耦和独立性至关重要，并确保系统仅因新的业务需求而需要更改，而不是上游内部数据模型更改。

The extent to which the internal data models are isolated from external consumers tends to become a point of contention in organizations moving toward event-driven microservices. Isolating the internal data model is essential for ensuring decoupling and independence of services and to ensure that systems need only change due to new business requirements, and not upstream internal data-model changes.

确保架构兼容性

Ensuring Schema Compatibility

模式序列化（以及验证）也可以内置到捕获工作流程中。这可以在事件写入发件箱表之前或之后执行。成功意味着事件可以在工作流程中继续进行，而失败可能需要手动干预以确定根本原因并避免数据丢失。

Schema serialization (and therefore, validation) can also be built into the capture workflow. This may be performed either before or after the event is written to the outbox table. Success means the event can be proceed in the workflow, whereas a failure may require manual intervention to determine the root cause and avoid data loss.

在将事务提交到发件箱表之前进行序列化可提供最有力的数据一致性保证。序列化失败将导致事务失败并回滚对内部表所做的任何更改，以确保发件箱表和内部表保持同步。该流程如图4-8所示。成功的验证将看到事件已序列化并准备好进行事件流发布。这种方法的主要优点是显着减少了内部状态和输出事件流之间的数据不一致。事件流数据被视为一等公民，发布正确的数据被认为与维护一致的内部状态同样重要。

Serializing prior to committing the transaction to the outbox table provides the strongest guarantee of data consistency. A serialization failure will cause the transaction to fail and roll back any changes made to the internal tables, ensuring that the outbox table and internal tables stay in sync. This process is shown in Figure 4-8. A successful validation will see the event serialized and ready for event stream publishing. The main advantage of this approach is that data inconsistencies between the internal state and the output event stream are significantly reduced. The event stream data is treated as a first-class citizen, and publishing correct data is considered just as important as maintaining consistent internal state.

图 4-8。在写入发件箱表之前序列化更改数据

在写入发件箱之前进行序列化还为您提供了对所有事务使用单个发件箱的选项。格式很简单，因为内容主要是具有目标输出事件流映射的序列化数据。如图 4-9所示。

Serializing before writing to the outbox also provides you with the option of using a single outbox for all transactions. The format is simple, as the content is predominantly serialized data with the target output event stream mapping. This is shown in Figure 4-9.

图 4-9。具有已验证和序列化事件的单个输出表（请注意用于路由目的的output_stream条目）

在发布之前进行序列化的一个缺点是，性能可能会因序列化开销而受到影响。这对于轻负载来说可能无关紧要，但对于较重负载可能会产生更重要的影响。您需要确保满足您的性能需求。

One drawback of serializing before publishing is that performance may suffer due to the serialization overhead. This may be inconsequential for light loads but could have more significant implications for heavier loads. You will need to ensure your performance needs remain met.

或者，可以在事件写入发件箱表后执行序列化，如图4-8所示。

Alternately, serialization can be performed after the event has been written to the outbox table, as is shown in Figure 4-8.

图 4-10。作为发布过程的一部分，写入发件箱表后序列化更改数据

使用此策略，您通常拥有独立的发件箱，每个域模型一个，映射到相应输出事件流的公共模式。发布者进程从发件箱读取未序列化的事件，并在将其生成到输出事件流之前尝试使用关联的模式对其进行序列化。图 4-11显示了多个发件箱的示例，一个用于用户实体，一个用于帐户实体。

With this strategy you typically have independent outboxes, one for each domain model, mapped to the public schema of the corresponding output event stream. The publisher process reads the unserialized event from the outbox and attempts to serialize it with the associated schema prior to producing it to the output event stream. Figure 4-11 shows an example of multiple outboxes, one for a User entity and one for an Account entity.

图 4-11。多个发件箱表（请注意，数据未序列化，这意味着它可能与输出事件流的架构不兼容）

序列化失败表示事件的数据不符合其定义的架构，因此无法发布。这就是写入后序列化选项变得更难以维护的地方，因为已经完成的事务会将不兼容的数据推送到发件箱表中，并且不能保证事务可以逆转。

A failure to serialize indicates that the data of the event does not comply with its defined schema and so cannot be published. This is where the serialization-after-write option becomes more difficult to maintain, as an already completed transaction will have pushed incompatible data into the outbox table, and there is no guarantee that the transaction can be reversed.

实际上，您的发件箱中通常会出现大量不可序列化的事件。很可能需要人工干预来尝试挽救一些数据，但解决问题将非常耗时且困难，甚至可能需要停机以防止出现其他问题。由于某些事件可能确实兼容并且已经发布，导致输出流中的事件顺序可能不正确，这一事实使情况变得更加复杂。

In reality, you will typically end up with a large number of unserializable events in your outbox. Human intervention will most likely be required to try to salvage some of the data, but resolving the issue will be time-consuming and difficult and may even require downtime to prevent additional issues. This is compounded by the fact that some events may indeed be compatible and have already been published, leading to possible incorrect ordering of events in output streams.

在写入之前进行验证和序列化可确保数据被视为一等公民，并保证输出事件流中的事件最终与源数据存储中的数据一致，同时还保持源内部的隔离数据模型。这是变更数据捕获解决方案可以提供的最有力的保证。

Validating and serializing before writing ensures that the data is being treated as a first-class citizen and offers a guarantee that events in the output event stream are eventually consistent with the data inside the source data store, while also preserving the isolation of the source’s internal data model. This is the strongest guarantee that a change-data capture solution can offer.

使用触发器捕获变更数据

Capturing Change-Data Using Triggers

触发器支持早于前面几节中讨论的许多审核、二进制日志和预写日志模式。许多旧的关系数据库使用触发器作为生成审计表的手段。顾名思义，触发器被设置为在特定条件下自动发生。如果失败，导致触发器执行的命令也会失败，从而确保更新原子性。

Trigger support predates many of the auditing, binlog, and write-ahead log patterns examined in the previous sections. Many older relational databases use triggers as a means of generating audit tables. As their name implies, triggers are set up to occur automatically on a particular condition. If it fails, the command that caused the trigger to execute also fails, ensuring update atomicity.

您可以使用AFTER触发器捕获对审计表的行级更改。例如，在任何INSERT、UPDATE、 或DELETE命令之后，触发器会将相应的行写入更改数据表。这确保了对特定表所做的更改得到相应的跟踪。

You can capture row-level changes to an audit table by using an AFTER trigger. For example, after any INSERT, UPDATE, or DELETE command, the trigger will write a corresponding row to the change-data table. This ensures that changes made to a specific table are tracked accordingly.

考虑图 4-12中所示的示例。用户数据被更新到用户表中，触发器在事件发生时捕获事件。请注意，触发器还捕获插入发生的时间以及供事件发布进程使用的自动递增序列 ID。

Consider the example shown in Figure 4-12. User data is being upserted to a user table, with a trigger capturing the events as they occur. Note that the trigger is also capturing the time at which the insertion occurred as well as an autoincrementing sequence ID for the event publisher process to use.

图 4-12。使用触发器捕获对用户表的更改

在触发器执行期间，您通常无法使用事件模式验证更改数据，尽管这并非不可能。一个主要问题是它可能根本不受支持，因为触发器在数据库本身内执行，并且许多触发器仅限于它们可以支持的语言形式。虽然 PostgreSQL 支持 C、Python 和 Perl，可用于编写用户定义的函数来执行模式验证，但许多其他数据库不提供多语言支持。最后，即使支持触发器，它也可能太昂贵。每个触发器独立触发，需要大量的开销来存储必要的数据、模式和验证逻辑，并且对于许多系统负载来说，成本太高。

You generally cannot validate the change-data with the event schema during the execution of a trigger, though it is not impossible. One main issue is that it may simply not be supported, as triggers execute within the database itself, and many are limited to the forms of language they can support. While PostgreSQL supports C, Python, and Perl, which may be used to write user-defined functions to perform schema validation, many other databases do not provide multilanguage support. Finally, even if a trigger is supported, it may simply be too expensive. Each trigger fires independently and requires a nontrivial amount of overhead to store the necessary data, schemas, and validation logic, and for many system loads the cost is too high.

图 4-13显示了上一个示例的延续。对变更数据执行事后验证和序列化，并将成功验证的数据生成到输出事件流。不成功的数据需要根据业务需求进行错误处理，但可能需要人工 干预。

Figure 4-13 shows a continuation of the previous example. After-the-fact validation and serialization is performed on the change-data, with successfully validated data produced to the output event stream. Unsuccessful data would need to be error-handled according to business requirements, but would likely require human intervention.

变更数据捕获表模式是内部表模式和输出事件流模式之间的桥梁。这三者之间的兼容性对于确保可以将数据生成到输出事件流至关重要。由于在触发器执行期间通常不执行输出模式验证，因此最好使更改数据表与输出事件模式的格式保持同步。

The change-data capture table schema is the bridge between the internal table schema and the output event stream schema. Compatibility among all three is essential for ensuring that data can be produced to the output event stream. Because output schema validation is typically not performed during trigger execution, it is best to keep the change-data table in sync with the format of the output event schema.

图 4-13。事后验证并生成输出事件流

话虽这么说，触发器在许多遗留系统中都可以很好地工作。根据定义，遗留系统倾向于使用旧技术；触发器已经存在了很长一段时间，并且很可能能够提供必要的变更数据捕获机制。访问和负载模式往往是明确定义且稳定的，因此可以准确估计添加触发的影响。最后，虽然模式验证不太可能在触发过程本身期间发生，但模式本身也同样不太可能发生改变，这仅仅是由于系统的遗留性质。仅当模式预计会频繁更改时，事后验证才成为一个问题。

That being said, triggers can work great in many legacy systems. Legacy systems tend to use, by definition, old technology; triggers have existed for a very long time and may very well be able to provide the necessary change-data capture mechanism. The access and load patterns tend to be well defined and stable, such that the impact of adding triggering can be accurately estimated. Finally, although schema validation is unlikely to occur during the triggering process itself, it may be equally unlikely that the schemas themselves are going to change, simply due to the legacy nature of the system. After-the-fact validation is only an issue if schemas are expected to change frequently.

处理查询和 CDC 日志模式的事后数据定义更改

Handling After-the-Fact Data Definition Changes for the Query and CDC Log Patterns

对于查询模式，可以在查询时获得模式，并且可以推断出事件模式。新的事件模式可以与输出事件流模式进行比较，模式兼容性规则用于允许或禁止发布事件数据。许多查询连接器都使用这种模式生成机制，例如Kafka Connect 框架提供的连接器。

For the query pattern, the schema can be obtained at query time, and an event schema can be inferred. The new event schema can be compared with the output event stream schema, with schema compatibility rules used to permit or prohibit publishing of the event data. This mechanism of schema generation is used by numerous query connectors, such as those provided with the Kafka Connect framework.

对于 CDC 日志模式，数据定义更新通常被捕获到 CDC 日志中自己的部分。这些更改需要从日志中提取并推断为代表数据集的模式。生成架构后，可以根据下游事件架构对其进行验证。然而，对此功能的支持是有限的。目前，Debezium 连接器仅支持 MySQL 的数据定义更改。

For the CDC log pattern, data definition updates are typically captured to their own part of the CDC log. These changes need to be extracted from the logs and inferred into a schema representative of the data set. Once the schema is generated, it can be validated against the downstream event schema. Support for this functionality, however, is limited. Currently, the Debezium connector supports only MySQL’s data definition changes.

处理变更数据表捕获模式的数据定义变更

Handling Data Definition Changes for Change-Data Table Capture Patterns

更改数据表充当输出事件流模式和内部状态模式之间的桥梁。应用程序的验证代码或数据库的触发函数中的任何不兼容都将阻止数据写入更改数据表，并将错误发送回堆栈。根据其模式兼容性规则，对变更数据捕获表进行的更改将需要与输出事件流兼容的模式演变。这涉及一个两步过程，可显着降低意外更改进入生产的机会。

The change-data table acts as a bridge between the output event stream schema and the internal state schema. Any incompatibilities in the application’s validation code or the database’s trigger function will prevent the data from being written to the change-data table, with the error sent back up the stack. Alterations made to the change-data capture table will require a schema evolution compatible with the output event stream, according to its schema compatibility rules. This involves a two-step process, which significantly reduces the chance of unintentional changes finding their way into production.

转换

Transformations

转换处理单个事件并发出零个或多个输出事件。正如您可能猜到的那样，转换提供了需要转换的大量业务逻辑操作。根据操作，事件可能需要重新分区（稍后将详细介绍）。常见的转换包括但不限于以下内容：

A transform processes a single event and emits zero or more output events. Transforms, as you might guess, provide the bulk of the business logic operations requiring transformations. Events may need to be repartitioned depending on the operations (more on this shortly). Common transformations include, but are not limited to, the following:

筛选

Filter

如果满足必要的标准，则传播该事件。发出零个或一个事件。

Propagate the event if it meets the necessary criteria. Emits zero or one events.

地图

Map

更改事件的键和/或值，仅发出一个事件。请注意，如果更改密钥，可能需要重新分区以确保数据局部性。

Changes the key and/or value of the event, emitting exactly one event. Note that if you change the key, you may need to repartition to ensure data locality.

地图值

MapValue

仅更改事件的值，而不更改键。只发出一个事件。不需要重新分区。

Change only the value of the event, not the key. Emits exactly one event. Repartitioning will not be required.

自定义变换

Custom transforms

应用自定义逻辑、查找状态，甚至与其他系统同步通信。

Apply custom logic, look up state, and even communicate with other systems synchronously.

分支和合并流

Branching and Merging Streams

消费者应用程序可能需要对事件流进行分支，即将逻辑运算符应用于事件，然后根据结果将其输出到新的流。一种相对常见的场景是使用事件的“消防水带”，并根据特定属性（例如，国家、时区、来源、产品或任意数量的特征）决定将它们路由到何处。第二种常见场景是将结果发送到不同的输出事件流，例如，在发生处理错误时将事件输出到死信流，而不是完全丢弃它们。

A consumer application may need to branch event streams—that is, apply a logical operator to an event and then output it to a new stream based on the result. One relatively common scenario is consuming a “firehose” of events and deciding where to route them based on particular properties (e.g., country, time zone, origin, product, or any number of features). A second common scenario is emitting results to different output event streams—for example, outputting events to a dead-letter stream in case of a processing error, instead of dropping them completely.

应用程序可能还需要合并流，其中使用来自多个输入流的事件，可能以某种有意义的方式进行处理，然后输出到单个输出流。在很多情况下，将多个流合并为一个流很重要，因为微服务通常会根据需要消耗尽可能多的输入流来实现其业务逻辑。第 6 章讨论如何以一致且可重现的顺序处理来自多个输入流的消费和处理事件。

Applications may also need to merge streams, where events from multiple input streams are consumed, possibly processed in some meaningful way, and then output to a single output stream. There aren’t too many scenarios where it’s important to merge multiple streams into just one, since it is common for microservices to consume from as many input streams as necessary to fulfill their business logic. Chapter 6 discusses how to handle consuming and processing events from multiple input streams in a consistent and reproducible order.

示例：重新分区事件流

Example: Repartitioning an Event Stream

假设有来自面向 Web 的端点的用户数据流。用户操作被转换为事件，事件的有效负载包含用户 ID 和其他任意事件数据，标记为x。

Suppose there is a stream of user data coming in from a web-facing endpoint. The user actions are converted into events, with the payload of the events containing both a user ID and other arbitrary event data, labeled x.

此状态的使用者有兴趣确保属于特定用户的所有数据都包含在同一分区中，无论源事件流如何分区。可以对该流重新分区以确保这种情况，如图5-2所示。

Consumers of this state are interested in ensuring that all of the data belonging to a particular user is contained within the same partition, regardless of how the source event stream is partitioned. This stream can be repartitioned to ensure this is the case, as shown in Figure 5-2.

将给定键的所有事件生成到单个分区中为数据局部性提供了基础。消费者只需要使用来自单个分区的事件来构建与该键相关的事件的完整图片。这使得消费者微服务能够扩展到多个实例，每个实例都从单个分区进行消费，同时维护与该密钥相关的所有事件的完整状态帐户。重新分区和数据局部性是大规模执行有状态处理的重要组成部分。

Producing all events for a given key into a single partition provides the basis for data locality. A consumer need only consume events from a single partition to build a complete picture of events pertaining to that key. This enables consumer microservices to scale up to many instances, each consuming from a single partition, while maintaining a complete stateful account of all events pertaining to that key. Repartitioning and data locality are essential parts of performing stateful processing at scale.

图 5-2。重新分区事件流

示例：对事件流进行共同分区

Example: Copartitioning an Event Stream

再次考虑图 5-2的重新分区示例。假设您现在需要将重新分区的用户事件流与用户实体流连接起来，并以相同的 ID 为关键。这些流的连接如图 5-3所示。

Consider again the repartition example of Figure 5-2. Say that you now need to join the repartitioned user event stream with a user entity stream, keyed on that same ID. These joining of these streams is shown in Figure 5-3.

两个流具有相同的分区计数，并且都使用相同的分区器算法进行分区。请注意，每个分区的键分布与其他流的分布相匹配，并且每个连接都由其自己的使用者实例执行。下一节将介绍如何将分区分配给微服务实例以利用共同分区的流，如本联接示例中所做的那样。

Both streams have the same partition count, and both have been partitioned using the same partitioner algorithm. Note that the key distribution of each partition matches the distribution of the other stream and that each join is performed by its own consumer instance. The next section covers how partitions are assigned to a microservice instance to leverage copartitioned streams, as was done in this join example.

图 5-3。共同分区的用户事件和用户实体流

使用分区分配器分配分区

Assigning Partitions with the Partition Assignor

处理大量数据通常需要消费者微服务的多个实例，无论是专用的流处理框架还是基本的生产者/消费者实现。分区分配器确保分区以平衡且公平的方式分配给处理实例。

Multiple instances of a consumer microservice are typically required for processing large volumes of data, whether it’s a dedicated stream-processing framework or a basic producer/consumer implementation. A partition assignor ensures that partitions are distributed to the processing instances in a balanced and equitable manner.

每当在消费者组中添加或删除新的消费者实例时，该分区分配器还负责重新分配分区。根据您的事件代理选择，该组件可能内置于消费者客户端中或在事件代理中维护。

This partition assignor is also responsible for reassigning partitions whenever new consumer instances are added or removed from the consumer group. Depending on your event broker selection, this component may be built into the consumer client or maintained within the event broker.

分配共同分区的分区

Assigning Copartitioned Partitions

分区分配者还负责确保满足任何共同分区要求。所有标记为共同分区的分区必须分配给同一个消费者实例。这可确保为给定的微服务实例分配正确的事件数据子集来执行其业务逻辑。最好让分区分配器实现检查事件流是否具有相等的分区计数，并在不相等时抛出异常。

The partition assignor is also responsible for ensuring that any copartitioning requirements are met. All partitions marked as copartitioned must be assigned to the same single consumer instance. This ensures that a given microservice instance will be assigned the correct subset of event data to perform its business logic. It is good practice to have the partition assignor implementation check to see that the event streams have an equal partition count and throw an exception on inequality.

分区分配策略

Partition Assignment Strategies

分区分配算法的目标是确保分区均匀分布在消费者实例上，假设消费者实例的处理能力相同。分区分配算法还可能有次要目标，例如减少重新平衡期间重新分配的分区数量。当您处理跨多个数据存储实例分片的物化状态时，这一点尤其重要，因为分区的重新分配可能会导致将来的更新进入错误的分片。第 7 章进一步探讨了有关内部状态存储的概念。

The goal of a partition assignment algorithm is to ensure that partitions are evenly distributed across the consumer instances, assuming that the consumer instances are equal in processing capabilities. A partition assignment algorithm may also have secondary goals, such as reducing the number of partitions reassigned during a rebalance. This is particularly important when you are dealing with materialized state sharded across multiple data store instances, as the reassignment of a partition can cause future updates to go to the wrong shard. Chapter 7 explores this concept further with regard to internal state stores.

有许多常见的分配分区策略。默认策略可能会根据您的框架或实现的不同而有所不同，但以下三种策略往往是最常用的。

There are a number of common strategies for assigning partitions. The default strategy may vary depending on your framework or implementation, but the following three tend to be the most commonly used.

循环分配

Round-robin assignment

所有分区都记录在一个列表中，并以循环方式分配给每个消费者实例。为共同分区的流保留一个单独的列表，以确保正确的共同分区分配。

All partitions are tallied into a list and assigned in a round-robin manner to each consumer instance. A separate list is kept for copartitioned streams to ensure proper copartitioned assignment.

图 5-4显示了两个使用者实例，每个实例都有自己的一组分配的分区。与 C1 的一组相比，C0 有两组共分区分区，因为分配都是在 C0 上开始和结束的。

Figure 5-4 shows two consumer instances, each with its own set of assigned partitions. C0 has two sets of copartitioned partitions compared to one for C1, since assignment both began and ended on C0.

图 5-4。两个消费者实例的循环分区分配

当给定消费者组的消费者实例数量增加时，应重新平衡分区分配，以在新添加的资源之间分散负载。图 5-5显示了添加两个以上消费者实例的效果 。

When the number of consumer instances for the given consumer group increases, partition assignments should be rebalanced to spread the load among the newly added resources. Figure 5-5 shows the effects of adding two more consumer instances.

图 5-5。四个消费者实例的循环分区分配

现在，C2 被分配了共同分区的 P2，以及流 A 的 P2。另一方面，C3 仅具有流 A 中的分区 P3，因为没有其他分区可供分配。添加任何其他实例将不会导致任何额外的 并行化。

C2 is now assigned the copartitioned P2s, as well as stream A’s P2. C3, on the other hand, only has partition P3 from stream A because there are no additional partitions to assign. Adding any further instances will not result in any additional parallelization.

同步分布式时间戳

Synchronizing Distributed Timestamps

物理学的一个基本限制是不能保证两个独立的系统具有完全相同的系统时钟时间。各种物理特性限制了系统时钟的精确度，例如底层时钟电路的材料容差、芯片工作温度的变化以及同步期间不一致的网络通信延迟。但是，可以建立几乎同步的本地系统时钟，并且最终足以满足大多数计算目的。

A fundamental limitation of physics is that two independent systems cannot be guaranteed to have precisely the same system-clock time. Various physical properties limit how precise system clocks can be, such as material tolerances in the underlying clock circuitry, variations in the operating temperature of the chip, and inconsistent network communication delays during synchronization. However, it is possible to establish local system clocks that are nearly in sync and end up being good enough for most computing purposes.

一致的时钟时间主要通过与网络时间协议 (NTP) 服务器同步来实现。亚马逊和谷歌等云服务提供商在各自地区提供冗余的卫星连接时钟和原子钟，以实现即时同步。

Consistent clock times are primarily accomplished by synchronizing with Network Time Protocol (NTP) servers. Cloud service providers such as Amazon and Google offer redundant satellite-connected and atomic clocks in their various regions for instant synchronization.

与局域网内的NTP服务器同步可以提供非常准确的本地系统时钟，15分钟后漂移仅几毫秒。根据 NTP 发明人 David Mills 的说法，在最佳情况下，通过更频繁的同步，可以将该时间减少到1 毫秒或更少，尽管间歇性网络问题可能会阻止在实践中达到该目标。开放互联网上的同步可能会导致更大的偏差，精度会降低到 +/- 100 毫秒的范围，如果您尝试重新同步来自全球不同地区的事件，这是一个需要考虑的因素。

Synchronization with NTP servers within a local area network can provide very accurate local system clocks, with a drift of only a few mS after 15 minutes. This can be reduced to 1 mS or less with more frequent synchronizations in best-case scenarios according to David Mills, NTP’s inventor, though intermittent network issues may prevent this target from being reached in practice. Synchronization across the open internet can result in much larger skews, with accuracy being reduced to ranges of +/− 100mS, and is a factor to be considered if you’re trying to resynchronize events from different areas of the globe.

NTP 同步也容易失败，因为网络中断、配置错误和瞬态问题可能会阻止实例同步。否则 NTP 服务器本身也可能变得不可靠或无响应。实例内的时钟可能会受到多租户问题的影响，就像共享底层硬件的基于 VM 的系统一样。

NTP synchronization is also prone to failure, as network outages, misconfiguration, and transient issues may prevent instances from synchronizing. The NTP servers themselves may also otherwise become unreliable or unresponsive. The clock within an instance may be affected by multitenancy issues, just as in VM-based systems sharing the underlying hardware.

对于绝大多数业务案例，频繁同步到 NTP 服务器可以为系统事件时间提供足够的一致性。NTP 服务器和 GPS 使用的改进已开始将 NTP 同步精度持续推向亚毫秒范围。分配为时间戳的创建时间和摄取时间值可以高度一致，但仍会出现轻微的乱序问题。本章稍后将介绍晚期事件的处理。

For the vast majority of business cases, frequent synchronization to NTP servers can provide sufficient consistency for system event time. Improvements to NTP servers and GPS usage have begun to push NTP synchronization accuracy consistently into the submillisecond range. The creation time and ingestion time values assigned as the timestamps can be highly consistent, though minor out-of-order issues will still occur. Handling of late events is covered later in this chapter.

使用时间戳事件进行处理

Processing with Timestamped Events

时间戳提供了一种以一致的时间顺序处理分布在多个事件流和分区中的事件的方法。许多用例要求您根据时间维护事件之间的顺序，并且无论事件流何时处理，都需要一致、可重现的结果。使用偏移量作为比较方法仅适用于单个事件流分区内的事件，而事件通常需要从多个不同的事件流进行处理。

Timestamps provide a way to process events distributed across multiple event streams and partitions in a consistent temporal order. Many use cases require you to maintain order between events based on time, and need consistent, reproducible results regardless of when the event stream is processed. Using offsets as a means of comparison works only for events within a single event stream partition, while events quite commonly need to be processed from multiple different event streams.

示例：处理多个分区时选择事件顺序

Example: Selecting order of events when processing multiple partitions

银行必须确保存款和取款事件流都按照正确的时间顺序进行处理。它保留了取款和存款的状态运行记录，当客户的账户余额低于 0 美元时会施加透支罚款。对于本示例，银行将存款放在一个事件流中，将取款放在另一个事件流中，如图6-2所示。

A bank must ensure that both deposit and withdrawal event streams are processed in the correct temporal order. It keeps a stateful running tally of withdrawals and deposits, applying an overdraft penalty when a client’s account balance drops below $0. For this example, the bank has its deposits in one event stream and its withdrawals in another stream, as shown in Figure 6-2.

图 6-2。事件应该按什么顺序处理？

一种简单的消费和处理记录的方法（可能是循环处理器）可能会首先处理 10 美元的存款，然后处理 25 美元的取款（产生负余额和透支罚款），最后处理 20 美元的存款。然而，这是不正确的，并且并不代表事件发生的时间顺序。此示例清楚地表明，您在使用和处理事件时必须考虑事件的时间戳。下一节将更详细地讨论这一点。

A naive approach to consuming and processing records, perhaps a round-robin processor, might process the $10 deposit first, the $25 withdrawal second (incurring a negative balance and overdraft penalties), and the $20 deposit third. This is incorrect, however, and does not represent the temporal order in which the events occurred. This example makes clear that you must consider the event’s timestamp when consuming and processing events. The next section discusses this in greater detail.

自定义事件调度程序

Custom Event Schedulers

一些流框架允许您实现自定义事件调度程序。例如，Apache Samza 允许您实现一个MessageChooser类，您可以在其中根据多种因素选择要处理的事件，例如某些事件流相对于其他事件流的优先级、挂钟时间、事件时间、事件元数据甚至内容在事件本身内。但是，在实现自己的事件调度程序时应该小心，因为许多自定义调度程序本质上是不确定的，如果需要重新处理，将无法生成可重现的结果。

Some streaming frameworks allow you to implement custom event schedulers. For example, Apache Samza lets you implement a MessageChooser class, where you select which event to process based on a number of factors, such as prioritization of certain event streams over others, the wall-clock time, event time, event metadata, and even content within the event itself. You should take care when implementing your own event scheduler, however, as many custom schedulers are nondeterministic in nature and won’t be able to generate reproducible results if reprocessing is required.

基于事件时间、处理时间和摄取时间的处理

Processing Based on Event Time, Processing Time, and Ingestion Time

基于时间的事件处理顺序要求您选择用作事件时间戳的时间点，如图6-1所示。可以在本地分配的事件时间和代理摄取时间之间进行选择。两个时间戳在生产-消费工作流程中仅出现一次，而挂钟和消费者摄取时间根据应用程序的执行时间而变化。

A time-based order of event processing requires you to select which point in time to use as the event’s timestamp, as per Figure 6-1. The choice is between the locally assigned event time and broker ingestion time. Both timestamps occur only once each in a produce-consume workflow, whereas the wall-clock and consumer ingestion time change depending on when the application is executed.

在大多数情况下，特别是当所有消费者和所有生产者都健康并且任何消费者组都没有事件积压时，所有四个时间点将彼此相差几秒之内。相反，对于处理历史事件的微服务，事件时间和消费者摄取时间将显着不同。

In most scenarios, particularly when all consumers and all producers are healthy and there is no event backlog for any consumer group, all four points in time will be within a few seconds of each other. Contrarily, for a microservice processing historic events, event time and consumer ingestion time will differ significantly.

为了最准确地描述现实世界中的事件，最好使用本地分配的事件时间，前提是您可以信赖它的准确性。如果生产者的时间戳不可靠（并且您无法修复它），那么下一个最佳选择是根据事件被引入事件代理的时间来设置时间戳。仅在事件代理和生产者无法通信的极少数情况下，真实事件时间与代理分配的事件时间之间可能存在很大的延迟。

For the most accurate depiction events in the real world, it is best to use the locally assigned event time provided you can rely on its accuracy. If the producer has unreliable timestamps (and you can’t fix it), your next best bet is to set the timestamps based on when the events are ingested into the event broker. It is only in rare cases where the event broker and the producer cannot communicate that there may be a substantial delay between the true event time and the one assigned by the broker.

对外部系统的请求-响应调用

Request-Response Calls to External Systems

从事件驱动拓扑内向外部系统发出的任何非事件驱动请求都可能会引入不确定的结果。根据定义，外部系统是在微服务外部进行管理的，这意味着在任何时间点，它们的内部状态以及对请求微服务的响应都可能不同。这是否重要完全取决于微服务的业务需求，并由您评估。

Any non-event-driven requests made to external systems from within an event-driven topology may introduce nondeterministic results. By definition, external systems are managed externally to the microservice, meaning that at any point in time their internal state and their responses to the requesting microservice may differ. Whether this is significant depends entirely on the business requirements of your microservice and is up to you to assess.

并行处理中的水印

Watermarks in Parallel Processing

水印对于协调多个独立消费者实例之间的事件时间特别有用。图 6-4显示了两个消费者实例的简单处理拓扑。每个消费者实例使用来自其自己分配的分区的事件，应用一个groupByKey函数，然后应用一个aggregate函数。这需要一个shuffle，其中具有相同键的所有事件都被发送到单个下游聚合实例。在这种情况下，实例0和实例1的事件根据key相互发送，以确保相同key的所有事件都在同一个分区中。

Watermarks are particularly useful for coordinating event time between multiple independent consumer instances. Figure 6-4 shows a simple processing topology of two consumer instances. Each consumer instance consumes events from its own assigned partition, applies a groupByKey function, followed by an aggregate function. This requires a shuffle, where all events with the same key are sent to a single downstream aggregate instance. In this case, events from instance 0 and instance 1 are sent to each other based on the key to ensure all events of the same key are in the same partition.

图 6-4。具有多个处理器的单个拓扑中的节点之间的水印传播

该图中有相当多的内容需要解开，所以让我们从头开始看一下。

There is a fair bit to unpack in this diagram, so let’s take a look at it from the start.

水印是在源函数处生成的，其中事件是从事件流分区消耗的。水印定义了该消费者的事件时间，并随着消费者节点的事件时间的增加而向下游传播（图 6-4中的#1 ）。

Watermarks are generated at the source function, where the events are consumed from the event stream partition. The watermarks define the event time at that consumer and are propagated downstream as the event time of the consumer node is incremented (#1 in Figure 6-4).

当水印到达时，下游节点更新其事件时间，并依次生成自己的新水印以向下游传播到其后继节点。具有多个输入的节点（例如 ）aggregate消耗来自多个上游输入的事件和水印。节点的事件时间是节点内部跟踪的所有输入源事件时间的最小值（图 6-4中的#2 ）。

Downstream nodes update their event time as the watermarks arrive, and in turn generate their own new watermark to propagate downstream to its successors. Nodes with multiple inputs, such as aggregate, consume events and watermarks from multiple upstream inputs. The node’s event time is the minimum of all of its input sources’ event times, which the node keeps track of internally (#2 in Figure 6-4).

在示例中，aggregate一旦来自 groupByKey-1 节点的水印到达（图6-4中的#3 ），两个节点的事件时间都会从 13 更新到 15。注意，水印不影响节点的事件调度；它只是通知节点它应该将时间戳早于水印的任何事件视为延迟。本章稍后将介绍处理延迟事件。

In the example, both aggregate nodes will have their event time updated from 13 to 15 once the watermark from the groupByKey-1 node arrives (#3 in Figure 6-4). Note that the watermark does not affect the event scheduling of the node; it simply notifies the node that it should consider any events with a timestamp earlier than the watermark to be considered late. Handling late events is covered later in this chapter.

Spark、Flink 和 Beam 等重量级处理框架需要专用的处理资源集群来大规模执行流处理。这是特别相关的，因为该集群还提供了跨任务通信和每个处理任务的集中协调的手段。重新分区事件（例如本示例中的groupByKey+aggregate操作）使用集群内部通信，而不是事件代理中的事件流。

Spark, Flink, and Beam, among other heavyweight processing frameworks, require a dedicated cluster of processing resources to perform stream processing at scale. This is particularly relevant because this cluster also provides the means for cross-task communications and centralized coordination of each processing task. Repartitioning events, such as with the groupByKey + aggregate operation in this example, use cluster-internal communications and not event streams in the event broker.

并行处理中的流时间

Stream Time in Parallel Processing

再次考虑图 6-4中相同的两个实例消费者示例，但这次使用 Kafka Streams 倡导的流时间方法（参见图 6-6）。一个显着的区别是 Kafka Streams 方法使用所谓的内部事件流将重新分区的事件发送回事件代理。然后，实例重新使用该流，所有重新分区的数据按键共同位于单个分区内。这在功能上与重量级集群内的 shuffle 机制相同，但不需要专用集群（注意：Kafka Streams 对微服务非常友好）。

Consider again the same two-instance consumer example from Figure 6-4, but this time with the stream time approach championed by Kafka Streams (see Figure 6-6). A notable difference is that the Kafka Streams approach sends the repartitioned events back to the event broker using what’s known as an internal event stream. This stream is then reconsumed by the instances, with all repartitioned data colocated by key within single partitions. This is functionally the same as the shuffle mechanism within the heavyweight cluster, but does not require a dedicated cluster (note: Kafka Streams is very microservice friendly).

图 6-6。通过重新分区事件流对事件进行洗牌

在此示例中，输入流中的事件根据其键重新分区，并写入重新分区事件流中。事件以 为关键A并B以 结束P1，而事件以 为关键X并Z以 结束P0。另请注意，每个事件的事件时间均已维护，并且不会被当前挂钟时间覆盖。回想一下，重新分区只能被视为现有事件数据的逻辑混洗。重写事件的事件时间会完全破坏原来的时间顺序。

In this example, events from the input stream are repartitioned according to their key and are written into the repartition event stream. The events keyed on A and B end up in P1, while the events keyed on X and Z end up in P0. Also note that the event time has been maintained for each event, and not overwritten by the current wall-clock time. Recall that the repartition should be treated only as a logical shuffling of existing event data. Rewriting the event time of an event would completely destroy the original temporal ordering.

请注意图中所示的子拓扑。由于重新分区事件流，处理拓扑实际上被减半，这意味着每个子拓扑上的工作可以并行完成。子拓扑1和3消费重新分区流并将事件组合在一起，同时子拓扑0和2生成重新分区事件。每个子拓扑都维护自己的流时间，因为两者都消耗独立的事件流。

Notice the subtopologies shown in the figure. Because of the repartition event stream, the processing topology is effectively cut in half, meaning that work on each subtopology can be done in parallel. Subtopologies 1 and 3 consume from the repartition stream and group events together, while subtopologies 0 and 2 produce the repartitioned events. Each subtopology maintains its own stream time, since both are consuming from independent event streams.

带有水印和直播时间的迟到活动

Late Events with Watermarks and Stream Time

考虑两个事件，一个事件的时间为t，另一个事件的时间为t '。事件t ′ 的时间戳早于事件t。

Consider two events, one with time t, the other with time t′. Event t′ has an earlier timestamp than event t.

水印

Watermarks

当事件t ' 在水印W(t)之后到达时，它被认为是迟到的。如何处理这个事件取决于具体节点。

The event t′ is considered late when it arrives after the watermark W(t). It is up to the specific node how to handle this event.

直播时间

Stream time

当事件t ' 在流时间增加超过t '之后到达时，事件 t ' 被视为迟到。如何处理该事件取决于子拓扑中的每个操作员。

The event t′ is considered late when it arrives after the stream time has been incremented past t′. It is up to each operator in the subtopology how to handle this event.

无序事件的原因和影响

Causes and Impacts of Out-of-Order Events

无序事件的发生有多种方式。

There are several ways that out-of-order events can occur.

多个生产者到多个分区

Multiple producers to multiple partitions

多个生产者写入多个输出分区可能会引入乱序事件。重新分区现有事件流是发生这种情况的一种方法。图6-8显示了两个消费者实例对两个分区的重新分区。在此场景中，源事件指示用户与哪个产品进行了交互。例如，Harry 与产品 ID12 和 ID77 进行了交互。假设数据分析师需要在用户 ID 上重新设置这些事件的密钥，以便他们可以对用户的参与度执行基于会话的分析。由此产生的输出流可能会出现一些无序事件。

Multiple producers writing to multiple output partitions can introduce out-of-order events. Repartitioning an existing event stream is one way in which this can happen. Figure 6-8 shows the repartitioning of two partitions by two consumer instances. In this scenario the source events indicate which product the user has interacted with. For instance, Harry has interacted with products ID12 and ID77. Say that a data analyst needs to rekey these events on the user ID, such that they can perform session-based analysis of the user’s engagements. The resultant output streams may end up with some out-of-order events.

图 6-8。通过重新分区事件流对事件进行洗牌

请注意，每个实例都维护自己的内部流时间，并且两个实例之间没有同步。这可能会导致时间偏差，从而产生无序事件，如图6-9所示。

Note that each instance maintains its own internal stream time and that there is no synchronization between the two instances. This can cause a time skew that produces out-of-order events, as shown in Figure 6-9.

图 6-9。通过重新分区事件流对事件进行洗牌

实例 0 在流时间上仅稍微领先于实例 1，但由于它们的流时间独立，时间t = 90 和t = 95的事件在重新分区的事件流中被视为无序。不平衡的分区大小、不平等的处理速率和大量事件积压会加剧此问题。这里的影响是，以前按顺序的事件数据现在是无序的，因此作为消费者，您不能依赖于每个事件流中持续增加的时间。

Instance 0 was only slightly ahead of instance 1 in stream time, but because of their independent stream times, the events of time t = 90 and t = 95 are considered out of order in the repartitioned event stream. This issue is exacerbated by unbalanced partition sizes, unequal processing rates, and large backlogs of events. The impact here is that the previously in-order event data is now out of order, and thus as a consumer you cannot depend on having consistently incrementing time in each of your event streams.

提示

单线程生产者在正常操作中不会创建乱序事件，除非它从乱序源获取数据。

A single-threaded producer will not create out-of-order events in normal operation unless it is sourcing its data from an out-of-order source.

由于只要检测到具有较高时间戳的事件，流时间就会增加，因此可能会出现大量事件由于重新排序而被视为延迟的情况。这可能会对处理产生影响，具体取决于消费者选择如何处理无序事件。

Since the stream time is incremented whenever an event with a higher timestamp is detected, it is possible to end up in a scenario where a large number of events are considered late due to reordering. This may have an effect on processing depending on how the consumers choose to handle out-of-order events.

时间敏感函数和窗口

Time-Sensitive Functions and Windowing

延迟事件主要关注基于时间的业务逻辑，例如聚合特定时间段内的事件或在经过特定时间段后触发事件。延迟事件是在业务逻辑已经完成该特定时间段的处理之后到达的事件。窗口函数是基于时间的业务逻辑的一个很好的例子。

Late events are predominantly the concern of time-based business logic, such as aggregating events in a particular time period or triggering an event after a certain period of time has passed. A late event is one that arrives after the business logic has already finished processing for that particular period of time. Windowing functions are an excellent example of time-based business logic.

窗口化意味着按时间将事件分组在一起。这对于具有相同键的事件特别有用，您希望了解该键的事件在该时间段内发生了什么。事件窗口分为三种主要类型，但再次强调，请务必检查您的流处理框架以获取更多信息。

Windowing means grouping events together by time. This is particularly useful for events with the same key, where you want to see what happened with events of that key in that period of time. There are three main types of event windows, but again, be sure to check your stream-processing framework for more information.

翻滚窗户

Tumbling windows

翻滚窗口是固定大小的窗口。前一个和后一个窗口不重叠。图 6-10显示了三个滚动窗口，每个窗口都在t、t + 1 等上对齐。这种窗口可以帮助回答诸如“产品使用高峰时间是什么时候？”之类的问题。

A tumbling window is a window of a fixed size. Previous and subsequent windows do not overlap. Figure 6-10 shows three tumbling windows, each aligned on t, t + 1, and so on. This sort of windowing can help answer questions such as “When is the peak hour for product usage?”

图 6-10。翻滚窗户

推拉窗

Sliding windows

滑动窗口具有固定的窗口大小和增量步长，称为窗口滑动。它必须仅反映窗口中当前事件的聚合。滑动窗口可以帮助回答诸如“过去一小时内有多少用户点击了我的产品？”之类的问题。图6-11显示了滑动窗口的示例，包括窗口的大小和向前滑动的量。

A sliding window has a fixed window size and incremental step known as the window slide. It must reflect only the aggregation of events currently in the window. A sliding window can help answer questions such as “How many users clicked on my product in the past hour?” Figure 6-11 shows an example of the sliding window, including the size of the window and the amount that it slides forward.

图 6-11。推拉窗

会话窗口

Session windows

会话窗口是动态调整大小的窗口。它根据由于不活动而超时而终止，并为超时后发生的任何活动启动新会话。图 6-12显示了会话窗口的示例，其中由于用户 C 不活动而出现会话间隙。这种窗口可以帮助回答诸如“用户在给定浏览会话中看什么？”之类的问题。

A session window is a dynamically sized window. It is terminated based on a timeout due to inactivity, with a new session started for any activity happening after the timeout. Figure 6-12 shows an example of session windows, with a session gap due to inactivity for user C. This sort of window can help answer questions such as “What does a user look at in a given browsing session?”

图 6-12。会话窗口

每个窗口函数都必须处理乱序事件。您必须决定等待任何无序事件多久，以免认为它们为时已晚而无法考虑。流处理中的基本问题之一是您永远无法确定已收到所有事件。等待无序事件可以起作用，但最终您的服务将需要放弃，因为它不能无限期地等待。其他需要考虑的因素包括要存储多少状态、延迟事件的可能性以及不使用延迟事件的业务影响。

Each of these window functions must deal with out-of-order events. You must decide how long to wait for any out-of-order events before deeming them too late for consideration. One of the fundamental issues in stream processing is that you can never be sure that you have received all of the events. Waiting for out-of-order events can work, but eventually your service will need to give up, as it cannot wait indefinitely. Other factors to consider include how much state to store, the likelihood of late events, and the business impact of not using the late events.

物化全球状态

Materializing Global State

全局状态存储是内部状态存储的一种特殊形式。全局状态存储不是仅具体化分配给它的分区，而是具体化给定事件流的所有分区的数据，从而为每个微服务实例提供事件数据的完整副本。图 7-4说明了全局物化状态和非全局物化状态之间的区别。

A global state store is a special form of the internal state store. Instead of materializing only the partitions assigned to it, a global state store materializes the data of all partitions for a given event stream, providing a complete copy of the event data to each microservice instance. Figure 7-4 illustrates the difference between global and nonglobal materialized state.

图 7-4。全局物化状态和非全局物化状态

当每个实例都需要完整的数据集并且往往包含小型、常用且很少更改的数据集时，全局状态存储非常有用。全局物化不能有效地用作事件驱动逻辑的驱动程序，因为每个微服务实例都拥有数据的完整副本，因此会产生重复的输出和不确定的结果。因此，最好为公共数据集查找和维度表保留全局具体化。

Global state stores are useful when a full data set is required by each instance, and tend to comprise small, commonly used, and seldom-changing data sets. Global materialization cannot be effectively used as the driver for event-driven logic, as each microservice instance possesses a full copy of the data and thus would produce duplicate outputs and nondeterministic results. For this reason, it is best to reserve global materialization for common data set lookup and dimension tables.

使用内部状态的优点

Advantages of Using Internal State

使用内部状态的缺点

Disadvantages of Using Internal State

内部状态的缩放和恢复

Scaling and Recovery of Internal State

从状态恢复的角度来看，将处理扩展到多个实例和恢复失败的实例是相同的过程。新的或恢复的实例需要具体化其拓扑定义的任何状态，然后才能开始处理新事件。最快的方法是为应用程序中具体化的每个有状态存储重新加载变更日志主题。

Scaling processing up to multiple instances and recovering a failed instance are identical processes from a state recovery perspective. The new or recovered instance needs to materialize any state defined by its topology before it can begin processing new events. The quickest way to do so is to reload the changelog topic for each stateful store materialized in the application.

使用热副本

Using hot replicas

虽然最常见的是每个分区只有一个物化状态副本，但可以通过一些仔细的状态管理来创建其他副本或由客户端框架直接利用。Apache Kafka通过简单的配置设置将此功能内置到其 Streams 框架中。此设置提供高度可用的状态存储，并使微服务能够以零停机时间容忍实例故障。

While it is most common to have only a single replica of materialized state for each partition, additional replicas can be created through some careful state management or leveraged directly by the client framework. Apache Kafka has this functionality built into its Streams framework via a simple configuration setting. This setting provides highly available state stores and enables the microservice to tolerate instance failures with zero downtime.

图 7-5显示了内部状态存储复制因子为 2 的三实例部署。每个有状态分区被物化两次，一次作为领导者，一次作为副本。每个副本必须管理自己的偏移量，以确保它与领导者副本的偏移量保持一致。实例 0 和实例 1 正在处理流 B 事件并将它们加入到共同分区的物化状态上。实例 1 和实例 2 还分别维护流 A-P0 和 A-P1 的热副本，否则实例 2 不处理任何其他事件。

Figure 7-5 shows a three-instance deployment with an internal state store replication factor of 2. Each stateful partition is materialized twice, once as the leader and once as a replica. Each replica must manage its own offsets to ensure that it is keeping up with the offsets of the leader replica. Instance 0 and instance 1 are processing stream B events and joining them on the copartitioned materialized state. Instance 1 and instance 2 are also maintaining hot replicas of stream A-P0 and A-P1, respectively, with instance 2 otherwise not processing any other events.

图 7-5。每个物化输入分区具有三个实例和两个热副本的流表连接

当领导者终止时，消费者组必须重新平衡分区的分配。分区分配器确定热副本的位置（它之前分配了所有分区并知道所有分区到实例的映射）并相应地重新分配分区。在图 7-6中，实例 1 已终止，其余微服务实例被迫重新平衡其分区分配。具有热副本的实例将优先获得分区所有权并立即恢复处理。分区分配器已选择实例 2 来恢复流 B-P1 的处理。

When the leader is terminated, the consumer group must rebalance the assignment of partitions. The partition assignor determines the location of the hot replica (it previously assigned all partitions and knows all partition-to-instance mappings) and reassigns the partitions accordingly. In Figure 7-6, instance 1 has terminated, and the remaining microservice instances are forced to rebalance their partition assignments. Instances with hot replicas are given priority to claim partition ownership and resume processing immediately. The partition assignor has selected instance 2 to resume processing of stream B-P1.

图 7-6。由于实例 1 终止而重新平衡

一旦处理恢复，必须根据变更日志构建新的热副本以维持最小副本数量。构建新的热副本并将其添加到剩余实例中，如图7-7所示。

Once processing has resumed, new hot replicas must be built from the changelog to maintain the minimum replica count. The new hot replicas are built and added to the remaining instances, as shown in Figure 7-7.

图 7-7。每个物化输入分区有两个实例和两个热副本的正常处理

笔记

热副本方法的主要权衡之一是使用额外的磁盘来维护副本，以换取因实例故障而减少的停机时间。

One of the main tradeoffs with a hot-replica approach is the use of additional disk to maintain the replicas in exchange for the reduction in downtime due to an instance failure.

外部国家的优势

Advantages of External State

外部国家的缺点

Drawbacks of External State

使用外部状态存储进行扩展和恢复

Scaling and Recovery with External State Stores

使用外部状态存储扩展和恢复微服务只需添加一个具有访问状态存储所需凭据的新实例。相反，底层状态存储的扩展和恢复完全取决于所选的技术，并且要复杂得多。

Scaling and recovery of microservices using an external state store simply require that you add a new instance with the necessary credentials to access the state store. In contrast, scaling and recovery of the underlying state store are dependent completely upon the selected technology and are much more complicated.

重申一下之前的观点，拥有一份可接受的外部数据服务列表以及如何正确管理、扩展、备份和恢复这些服务的指南对于为开发人员提供可持续的前进方向至关重要。不幸的是，状态存储技术的数量过多，实际上不可能在本书中进行讨论。相反，我将简单地将构建状态的策略概括为三种主要技术：从源流重建、使用更改日志和创建 快照。

To reiterate an earlier point, having a list of acceptable external data services with guides on how to properly manage, scale, back up, and restore them is essential for providing developers a sustainable way forward. Unfortunately, the number of state store technologies is prohibitively large and effectively impossible to discuss in this book. Instead, I’ll simply generalize the strategies of building state into three main techniques: rebuilding from the source streams, using changelogs, and creating snapshots.

重建

Rebuilding

重建微服务的状态存储通常是更新应用程序内部状态的最常见方法。微服务首先停止，消费者输入流偏移重置到开头。任何中间状态，例如存储在变更日志中或位于外部状态存储中的状态，都必须删除。最后，启动新版本的微服务，并在从输入事件流读回事件时重建状态。这种方法确保状态完全按照新业务逻辑的指定构建。所有新的输出事件也会被创建并向下游传播到订阅消费者。这些不被视为重复事件，因为业务逻辑和输出格式可能已更改，并且这些更改必须向下游传播。

Rebuilding the microservice’s state stores is typically the most common method of updating the internal state of an application. The microservice is first stopped, with the consumer input stream offsets reset to the beginning. Any intermediate state, such as that stored in a changelog or located in the external state store, must be deleted. Lastly, the new version of the microservice is started up, and the state is rebuilt as events are read back in from the input event streams. This approach ensures that the state is built exactly as specified by the new business logic. All new output events are also created and propagated downstream to the subscribing consumers. These are not considered duplicate events, as the business logic and output format may have changed, and these changes must be propagated downstream.

重建状态要求所有必要的输入事件流事件仍然存在，特别是需要状态和聚合具体化的任何事件。如果您的应用程序严重依赖一组输入数据，则必须确保此类源数据在微服务实现的数据存储之外随时可用。

Rebuilding state requires that all necessary input event stream events still exist, particularly anything that requires materialization of state and aggregations. If your application is critically reliant upon a set of input data, you must ensure that such source data is readily available outside of your microservice implementation’s data store.

最后，某些业务需求绝对要求您从一开始就重新处理数据，例如提取仅存在于输入事件中的字段。除了重播输入事件之外，您无法通过任何其他方式获取此数据，此时重建状态存储是唯一可行的选择。

Finally, some business requirements absolutely require you to reprocess data from the beginning of time, such as those that extract fields that are present only in the input events. You cannot obtain this data in any other way than replaying the input events, at which point rebuilding the state store is the only viable option.

迁移

Migrating

与更改的影响相比，大型状态存储可能需要很长时间才能重建，或者可能导致极其昂贵的数据传输成本。例如，考虑业务需求更改，其中要向微服务的输出事件流添加一个附加但可选的字段。此更改可能需要您向微服务的状态存储添加另一个列或字段。但是，企业可能不需要重新处理旧数据，而只想将逻辑应用于未来的新输入事件。对于由关系数据库支持的状态存储，您只需要更新业务逻辑以及关联的表定义。您可以简单地插入一个具有可为空默认值的新列，经过一系列快速测试后，您可以重新部署该应用程序。

Large state stores can take a long time to rebuild or can result in prohibitively expensive data transfer costs when compared to the impact of the change. For instance, consider a business requirement change where an additional, yet optional, field is to be added to a microservice’s output event stream. This change may require you to add another column or field to the microservice’s state store. However, it could be that the business has no need to reprocess older data and wants only to apply the logic to new input events going forward. For a state store backed by a relational database, you’d just need to update the business logic alongside the associated table definition. You can perform a simple insertion of a new column with a nullable default, and after a quick series of tests you can redeploy the application.

当业务需求和要更改的数据更加复杂时，迁移的风险就会变得更大。复杂的迁移很容易出错，并且与完全重建数据存储的结果相比，可能会产生不正确的结果。数据库迁移逻辑不是业务逻辑的一部分，因此可能会引入不一致，而这些不一致在应用程序的完全重建期间不会出现。如果在测试期间未发现这些类型的迁移错误，则很难检测到，并且可能导致数据不一致。当采用基于迁移的方法时，请务必执行严格的测试并使用代表性测试数据集将该方法与基于重建的方法进行比较。

Migrations become much riskier when the business need and the data being changed are more complex. Complex migrations are error-prone and can yield results that are incorrect compared to the results of a complete rebuild of the data store. The database migration logic is not a part of the business logic, so inconsistencies can be introduced that would otherwise not arise during a full rebuild of the application. These types of migration errors can be hard to detect if not caught during testing and can lead to inconsistent data. When following a migration-based approach, be sure to perform strict testing and use representative test data sets to compare that approach with a rebuild-based one.

示例：股票会计服务

Example: Stock Accounting Service

库存会计服务负责在任何给定商品的库存不足时发出通知事件。微服务必须根据随着时间推移进行的一系列增减，将每种产品的当前可用库存拼凑起来。向客户出售商品、因损坏而丢失商品以及因被盗而丢失商品都是减少库存的事件，而接收货物和接受客户退货会增加库存。为了简单起见，在本示例中这些事件显示在同一事件流中，如图7-8所示。

The stock accounting service is responsible for issuing a notification event when stock of any given item is low. The microservice must piece together the current stock available for each product based on a chain of additions and subtractions made over time. Selling items to customers, losing items to damage, and losing items to theft are all events that reduce stock, whereas receiving shipments and accepting customer returns increase it. These events are shown in the same event stream for simplicity in this example, as illustrated in Figure 7-8.

图 7-8。简单的股票会计服务

这种股票会计服务非常简单。它根据事件流变化计算当前运行的库存总量，并将其存储在数据存储中。业务逻辑根据阈值进行过滤，并决定是否向库存管理发出有关低库存或超卖库存的通知。必须确保每个输入事件有效地应用于聚合状态一次，因为多次应用它是不正确的，就像根本不应用它一样。一旦处理发挥作用，这就是有效的地方。

This stock accounting service is quite simple. It calculates the current running total of stock based off of the event stream changes and stores it in its data store. The business logic filters on a threshold value and decides whether to issue a notification to stock management about low or oversold stock. It must be ensured that each input event is applied effectively once to the aggregated state, as applying it more than once is incorrect, as is not applying it at all. This is where effectively once processing comes into play.

一次有效地处理客户经纪商交易

Effectively Once Processing with Client-Broker Transactions

任何支持事务的事件代理都可以有效地促进处理。通过这种方法，任何输出事件、对由变更日志支持的内部状态进行的更新以及消费者偏移量的增量都被包装在单个原子事务中。仅当所有这三个更新都存储在代理中其自己的特定事件流中时，这才有可能。偏移量更新、变更日志更新和输出事件在单个事务中以原子方式提交，如图7-9所示。

Effectively once processing can be facilitated by any event broker that supports transactions. With this approach, any output events, updates made to internal state backed by a changelog, and the incrementing of the consumer offsets are wrapped together within a single atomic transaction. This is possible only if all three of these updates are stored within their own specific event stream in the broker. The offset update, the changelog update, and the output event are committed atomically within a single transaction as shown in Figure 7-9.

图 7-9。客户-经纪商交易——提交偏移量和变更日志

生产者客户端和事件代理之间的原子事务会将所有事件发布到其相应的事件流。在生产者永久失败的情况下，如图7-10所示，代理将确保事务中的任何事件都不会提交。事件流使用者通常不会处理未提交事务中的事件。消费者必须遵守偏移顺序，因此它将阻塞，等待事务完成，然后继续处理事件。在出现暂时性错误的情况下，生产者可以简单地重试提交其事务，因为它是幂等操作。

The atomic transaction between the producer client and the event broker will publish all events to their corresponding event streams. In the case of permanent failure by the producer, as in Figure 7-10, the broker will ensure that none of the events in the transaction is committed. Event stream consumers typically abstain from processing events that are in uncommitted transactions. The consumer must respect offset order, and so it will block, wait for the transaction to complete, and then proceed to process the event. In the case of transient errors, the producer can simply retry committing its transaction, as it is an idempotent operation.

图 7-10。客户端代理事务提交失败

如果生产者在事务期间遇到致命异常，则可以通过从更改日志中恢复来重建其替换实例，如图7-11所示。输入事件流的消费者组偏移量也会根据偏移事件流中存储的最后一个已知的良好位置进行重置。

In the case that the producer suffers a fatal exception during a transaction, its replacement instance can simply be rebuilt by restoring from the changelogs as shown in Figure 7-11. The consumer group offsets of the input event streams are also reset according to the last known good position stored in the offset event stream.

一旦生产者恢复，新的事务就可以开始，并且所有先前未完成的事务都会失败并由事件代理清理。事务机制会根据代理实现的不同而有所不同，因此请确保熟悉您正在使用的机制。

New transactions can begin once the producer is recovered, and all previous incomplete transactions are failed and cleaned up by the event broker. The transactional mechanisms will vary to some extent depending on the broker implementation, so make sure to familiarize yourself with the one you are using.

图 7-11。使用更改日志和之前的偏移量从代理恢复状态

无需客户经纪人交易即可有效处理一次

Effectively Once Processing Without Client-Broker Transactions

对于不支持客户端代理事务的实现，也可以有效地一次处理事件，尽管它需要更多的工作并仔细考虑重复事件。首先，如果上游服务不能有效提供一次事件生产保证，那么就有可能产生重复记录。需要识别并过滤掉上游流程创建的任何重复事件。其次，状态和偏移管理需要在本地事务中更新以确保事件处理仅应用于系统状态一次。通过遵循此策略，客户端可以确信其处理器生成的内部状态与输入事件流的逻辑叙述一致。让我们更详细地看看这些步骤。

Effectively once processing of events is also possible for implementations that do not support client-broker transactions, though it requires more work and a careful consideration of duplicate events. First, if upstream services are not able to provide effectively once event production guarantees, then it is possible that they may produce duplicate records. Any duplicate events created by upstream processes need to be identified and filtered out. Second, state and offset management need to be updated in a local transaction to ensure that the event processing is applied only once to the system state. By following this strategy, clients can be assured that the internal state generated by their processor is consistent with the logical narrative of the input event streams. Let’s take a look at these steps in more detail.

防止重复

Guarding against duplicates

任何有效的一次消费者必须识别并丢弃重复项，执行幂等操作，或者从具有幂等生产者的事件流中进行消费。幂等操作并不适用于所有业务案例，如果没有幂等生产，您必须找到一种方法来保护业务逻辑免受重复事件的影响。这可能是一项昂贵的工作，因为它要求每个消费者维护先前处理的重复数据删除 ID 的状态存储。根据事件量以及应用程序必须防范的偏移量或时间范围，存储可能会变得非常大。

Any effectively once consumer must either identify and discard duplicates, perform idempotent operations, or consume from event streams that have idempotent producers. Idempotent operations are not possible for all business cases, and without idempotent production you must find a way to guard your business logic against duplicate events. This can be an expensive endeavor, as it requires that each consumer maintain a state store of previously processed dedupe IDs. The store can grow very large depending on the volume of events and the offset or time range that the application must guard against.

完美的重复数据删除要求每个使用者无限期地维护对已处理的每个重复数据删除 ID 的查找，但如果尝试防止范围过大，时间和空间要求可能会变得非常昂贵。实际上，重复数据删除通常仅针对特定的滚动时间窗口或偏移窗口执行，作为尽力而为的尝试。

Perfect deduplication requires that each consumer indefinitely maintain a lookup of each dedupe ID already processed, but time and space requirements can become prohibitively expensive if an attempt is made to guard against too large a range. In practice, deduplication is generally only performed for a specific rolling time-window or offset-window as a best-effort attempt.

提示

通过使用生存时间 (TTL)、最大缓存大小和定期删除来保持重复数据删除存储较小。所需的具体设置将根据应用程序对重复的敏感性以及重复发生的影响而有所不同。

Keep deduplication stores small by using time-to-live (TTL), a maximum cache size, and periodic deletions. The specific settings needed will vary depending on the sensitivity of your application to duplicates and the impact of duplicates occurring.

应仅在单个事件流分区内尝试重复数据删除，因为分区之间的重复数据删除将非常昂贵。与非键控事件相比，键控事件具有额外的优势，因为它们将始终分布到同一分区。

Deduplication should be attempted only within a single event stream partition, as deduplication between partitions will be prohibitively expensive. Keyed events have an added benefit over unkeyed events, since they will consistently be distributed to the same partition.

图 7-12显示了正在运行的重复数据删除存储。在此图中，您可以看到事件在传递到实际业务逻辑之前所经历的工作流程。在本例中，TTL任意设置为8000秒，但实际中需要根据业务需求来确定。

Figure 7-12 shows a deduplication store in action. In this figure you can see the workflow that an event goes through before being passed off to the actual business logic. In this example the TTL is arbitrarily set to 8,000 seconds, but in practice would need to be established based on business requirements.

图 7-12。使用持久状态进行重复数据删除

提示

重复数据删除存储中使用最大高速缓存大小来限制所维护的事件数量，特别是在重新处理期间。

A maximum cache size is used in the deduplication store to limit the number of events maintained, particularly during reprocessing.

请注意，您有责任维护重复数据删除表的持久备份，就像任何其他物化表一样。如果发生故障，必须在恢复处理新事件之前重建表。

Note that you are responsible for maintaining durable backups of the deduplication table, just as for any other materialized table. In the case of a failure, the table must be rebuilt prior to resuming the processing of new events.

保持一致的状态

Maintaining consistent state

微服务可以利用其状态存储的事务功能而不是事件代理来有效执行一次处理。这需要将消费者组的偏移管理从事件代理转移到数据服务中，从而允许单个有状态事务以原子方式更新状态和输入偏移。对状态所做的任何更改与对消费者偏移量所做的更改完全一致，从而保持服务内的一致性。

A microservice can leverage the transactional capabilities of its state store instead of the event broker to perform effectively once processing. This requires moving the consumer group’s offset management from the event broker and into the data service, allowing a single stateful transaction to atomically update both the state and input offsets. Any changes made to the state coincide completely with those made to the consumer offsets, which maintains consistency within the service.

如果出现服务故障，例如提交数据服务超时，微服务可以简单地放弃事务并恢复到最后一个已知的良好状态。所有消耗都会停止，直到数据服务做出响应，此时消耗将从最后一个已知的良好偏移量恢复。通过使偏移量的官方记录与数据服务中的数据保持同步，您可以获得服务可以恢复的一致状态视图。该过程如图7-13、7-14和7-15所示。

In the case of a service failure, such as a timeout when committing to the data service, the microservice can simply abandon to transaction and revert to the last known good state. All consumption is halted until the data service is responsive, at which point consumption is restored from the last known good offset. By keeping the official record of offsets synchronized with the data in the data service, you have a consistent view of state that the service can recover from. This process is illustrated in Figures 7-13, 7-14, and 7-15.

图 7-13。事件的正常事务处理

图 7-14。事务处理中发生故障

图 7-15。状态恢复过程中偏移量的恢复

请注意，此方法使您的处理器可以有效地进行一次处理，但不能有效地一次进行事件生成。此服务生成的任何事件都受到至少一次生成的限制，因为事件的非交易性客户端代理生成可能会创建重复项。

Note that this approach gives your processor effectively once processing, but not effectively once event production. Any events produced by this service are subject to the limitations of at-least-once production, as the nontransactional client-broker production of events is subject to the creation of duplicates.

如果您必须在更新状态存储的同时以事务方式生成事件，请参阅第 4 章。使用更改数据表可以为状态存储提供最终一致、有效的一次更新，并为输出事件流提供至少一次生成。

If you must have events transactionally produced alongside updates to the state store, refer to Chapter 4. Using a change-data table can provide eventually consistent, effectively once updates to the state store with at-least-once production to the output event stream.

一个简单的事件驱动编排示例

A Simple Event-Driven Choreography Example

图 8-1显示了精心设计的工作流的输出，其中服务 A 直接馈送到服务 B，服务 B 又馈送到服务 C。在这种特殊情况下，您可以推断服务具有 A → B → C 的依赖工作流服务C的输出表示整个工作流程的结果。

Figure 8-1 shows the output of a choreographed workflow in which service A feeds directly into service B, which in turn feeds into service C. In this particular case, you can infer that the services have a dependent workflow of A → B → C. The output of service C indicates the result of the workflow as a whole.

图 8-1。简单的事件驱动的精心设计的工作流程

现在，假设需要重新安排工作流程，使得服务 C 中的业务操作必须先于服务 B 中的业务操作执行，如图8-2所示。

Now, say that the workflow needs to be rearranged such that the business actions in service C must be performed before those in service B, as shown in Figure 8-2.

图 8-2。简单的事件驱动的精心设计的 工作流程所需的业务变更

服务 C 和 B 都必须经过编辑才能分别从流 1 和 2 中使用。流中数据的格式可能不再适合新工作流程的需要，需要进行破坏性架构更改，这可能会显着影响事件流 2 的其他使用者（未显示）。可能需要为事件创建全新的事件架构流 2，事件流中的旧数据移植为新格式、删除或保留在原处。最后，在交换拓扑之前，必须确保服务 A、B、C 的所有输出事件都已完成处理，以免某些事件处理不完全而处于不一致的状态。

Both services C and B must be edited to consume from streams 1 and 2, respectively. The format of the data within the streams may no longer suit the needs of the new workflow, requiring breaking schema changes that may significantly affect other consumers (not shown) of event stream 2. A whole new event schema may need to be created for event stream 2, with the old data in the event stream ported over to the new format, deleted, or left in place. Finally, you must ensure that processing has completed for all of the output events of services A, B, and C before swapping the topology around, lest some events be incompletely processed and left in an inconsistent state.

创建和修改精心设计的工作流程

Creating and Modifying a Choreographed Workflow

虽然编排允许在工作流程末尾简单添加新步骤，但在中间插入步骤或更改工作流程的顺序可能会出现问题。在工作流上下文之外，服务之间的关系也可能难以理解，随着工作流中服务数量的增加，这一挑战会加剧。精心设计的工作流程可能很脆弱，特别是当业务功能跨多个微服务实例时。通过仔细尊重服务的有界上下文并确保完整的业务功能仍然位于单个服务的本地，可以缓解这种情况。然而，即使正确实现，小的业务逻辑更改也可能需要您修改或重写大量服务，尤其是那些更改工作流本身顺序的服务。

While choreography allows for simple addition of new steps at the end of the workflow, it may be problematic to insert steps into the middle or to change the order of the workflow. The relationships between the services may also be difficult to understand outside the context of the workflow, a challenge that is exacerbated as the number of services in the workflow increases. Choreographed workflows can be brittle, particularly when business functions cross multiple microservice instances. This can be mitigated by carefully respecting the bounded contexts of services and ensuring that full business functionality remains local to a single service. However, even when correctly implemented, small business logic changes may require you to modify or rewrite numerous services, especially those that change the order of the workflow itself.

简单的直接调用编排示例

A Simple Direct-Call Orchestration Example

编排还可以使用请求-响应模式，其中编排器同步调用微服务的 API 并等待结果响应。除了直接调用的替代之外，图 8-4中所示的拓扑与图 8-3中的拓扑几乎相同。

Orchestration can also use a request-response pattern, where the orchestrator synchronously calls the microservice’s API and awaits a response for results. The topology shown in Figure 8-4 is nearly identical to the one in Figure 8-3, aside from substitution of direct calls.

图 8-4。简单的直接调用协调工作流程

直接呼叫服务的正常优点和限制也适用于此。话虽这么说，这种模式对于使用功能即服务解决方案实现工作流特别有用（请参阅第 9 章）。

The normal benefits and restrictions of direct-call services apply here as well. That being said, this pattern is particularly useful for implementing workflows with Function-as-a-Service solutions (see Chapter 9).

比较直接调用和事件驱动的编排

Comparing Direct-Call and Event-Driven Orchestration

仔细观察时，直接调用和事件驱动的编排工作流程非常相似。例如，事件驱动系统似乎实际上只是一个请求-响应系统，在这些简单的示例中，它确实如此。但是，当您缩小一点时，在选择使用哪个选项时需要考虑许多因素。

Direct-call and event-driven orchestration workflows are fairly similar when examined close-up. For instance, it might seem that the event-driven system is really just a request-response system, and in these simple examples, it certainly is. But when you zoom out a bit, there are a number of factors to consider when choosing which option to use.

事件驱动的工作流程：

Event-driven workflows:

• 可以使用与其他事件驱动微服务相同的 I/O 监控工具和滞后扩展功能

• Can use the same I/O monitoring tooling and lag scaling functionality as other event-driven microservices

• 允许事件流仍由其他服务使用，包括编排之外的服务

• Allow event streams to still be consumed by other services, including those outside of the orchestration

• 通常更持久，因为协调器和依赖服务都通过事件代理与彼此的间歇性故障隔离

• Are generally more durable, as both the orchestrator and the dependent services are isolated from each other’s intermittent failures via the event broker

• 具有内置的失败重试机制，因为事件可以保留在事件流中以便重试

• Have a built-in retry mechanism for failures, as events can remain in the event stream for retrying

直接呼叫工作流程：

Direct-call workflows:

• 通常更快，因为事件流的生成和消费没有开销

• Are generally faster, as there’s no overhead in producing to and consuming from event streams

• 存在可能需要由协调器管理的间歇性连接问题

• Have intermittent connectivity issues that may need to be managed by the orchestrator

总而言之，如果所有依赖服务都在其 SLA 范围内运行，直接调用（又称为同步请求响应工作流程）通常比事件驱动的工作流程更快。当需要非常快速的响应时，例如在实时操作中，它们往往效果很好。同时，事件驱动的工作流程具有更持久的 I/O 流，产生更慢但更稳健的操作，并且特别擅长处理间歇性故障。

To summarize, direct-call aka synchronous request-response workflows are generally faster than event-driven ones, provided all dependent services are operating within their SLAs. They tend to work well when very quick responses are required, such as in real-time operations. Meanwhile, event-driven workflows have more durable I/O streams, producing slower but more robust operations, and are particularly good at handling intermittent failures.

请记住，有很多机会混合和匹配这两个选项。例如，编排工作流程可能主要是事件驱动的，但要求直接对外部 API 或预先存在的服务进行请求-响应调用。将这两个选项混合在一起时，请确保按预期处理每个服务的故障模式。

Keep in mind that there’s quite a lot of opportunity to mix and match these two options. For example, an orchestration workflow may be predominantly event-driven, but require that a request-response call be made directly to an external API or preexisting service. When mixing these two options together, make sure that each service’s failure modes are handled as expected.

创建和修改编排工作流程

Creating and Modifying an Orchestration Workflow

编排器可以通过具体化每个传入和传出事件流以及响应请求结果来跟踪工作流中的事件。工作流本身仅在编排服务中定义，允许通过单点更改来更改工作流。在许多情况下，您可以将更改合并到工作流程中，而不会中断部分处理的事件。

The orchestrator can keep track of events in the workflow by materializing each of the incoming and outgoing event streams and response-request results. The workflow itself is defined solely within the orchestration service, allowing a single point of change for altering the workflow. In many cases, you can incorporate changes to a workflow without disrupting partially processed events.

编排导致服务之间的紧密耦合。必须显式定义协调器和依赖的工作人员服务之间的关系。

Orchestration results in a tight coupling between services. The relationship between the orchestrator and the dependent worker services must be explicitly defined.

确保编排器仅负责编排业务工作流程非常重要。一个常见的反模式是创建一个单一的“上帝”服务，向许多弱的 Minion 服务发出细粒度的命令。这种反模式在编排器和工作人员服务之间传播工作流业务逻辑，导致封装不良、定义不明确的有界上下文，并且难以将工作流微服务的所有权扩展到单个团队之外。编排器应将全部责任委托给依赖的工作人员服务，以最大程度地减少其执行的业务逻辑量。

It is important to ensure that the orchestrator is responsible only for orchestrating the business workflow. A common anti-pattern is creating a single “God” service that issues granular commands to many weak minion services. This anti-pattern spreads workflow business logic between the orchestrator and the worker services, making for poor encapsulation, ill-defined bounded contexts, and difficulty in scaling ownership of the workflow microservices beyond a single team. The orchestrator should delegate full responsibility to the dependent worker services to minimize the amount of business logic it performs.

监控编排工作流程

Monitoring the Orchestration Workflow

您可以通过查询具体化状态来了解编排工作流程，因此可以轻松查看任何特定事件的进度以及工作流程中可能出现的任何问题。您可以在协调器级别实施监控和日志记录，以检测导致工作流失败的任何事件。

You gain visibility into the orchestration workflow by querying the materialized state, so it’s easy to see the progress of any particular event and any issues that may have arisen in the workflow. You can implement monitoring and logging at the orchestrator level to detect any events that result in workflow failures.

精心设计的事务：Saga 模式

Choreographed Transactions: The Saga Pattern

具有精心设计的工作流程的分布式事务可能是复杂的事务，因为每个服务都需要能够在发生故障时回滚更改。这会在原本松散耦合的服务之间产生强耦合，并可能导致一些不相关的服务彼此严格依赖。

Distributed transactions with choreographed workflows can be complex affairs, as each service needs to be able to roll back changes in the event of a failure. This creates strong coupling between otherwise loosely coupled services and can result in some unrelated services having strict dependencies on one another.

精心设计的传奇模式适用于简单的分布式事务，特别是那些具有强烈的工作流排序要求且不太可能随时间变化的事务。监视精心设计的事务的进度可能很困难，因为它需要每个参与事件流的完全具体化，就像在精心设计的方法中一样。

The choreographed saga pattern is suitable for simple distributed transactions, particularly those with strong workflow ordering requirements that are unlikely to change over time. Monitoring the progress of a choreographed transaction can be difficult, because it requires a full materialization of each participating event stream, as in the orchestrated approach.

编排示例

Choreography Example

继续前面精心设计的工作流示例，考虑一系列微服务 A、B、C。服务 A 的输入事件流启动事务，服务 A、B 和 C 的工作完全完成，或者因此被取消并回滚。链中任何步骤的失败都会中止事务并开始回滚。最终的工作流程如图 8-5所示。

Continuing with the previous choreographed workflow example, consider the series of microservices A, B, C. The input event stream to service A kicks off a transaction, with the work of services A, B, and C being fully completed, or consequently canceled and rolled back. A failure at any step in the chain aborts the transaction and begins the rollback. The resultant workflow is shown in Figure 8-5.

图 8-5。精心设计的交易成功

现在假设服务 C 无法完成其事务部分。它现在必须通过发出事件或响应先前服务的请求来反转工作流程。服务 B 和 A 必须按顺序恢复其事务部分，如图8-6所示。

Suppose now that service C is unable to complete its part of the transaction. It must now reverse the workflow, either by issuing events or by responding to the previous service’s request. Services B and A must revert their portion of the transaction, in order, as shown in Figure 8-6.

图 8-6。精心设计的事务失败并回滚

服务 A（输入事件的原始使用者）现在必须决定如何处理失败的事务结果。前两张图中已经明显出现了一个奇怪的情况。成功交易的状态来自微服务 C 的输出。然而，中止交易的状态来自微服务 A，因此消费者需要监听 C 的输出和来自 A 的失败交易流才能获取已完成交易的完整情况。

Service A, the original consumer of the input event, must now decide what to do with the failed transaction results. A curious situation is already evident in the preceding two figures. The status of a successful transaction comes from the output of microservice C. However, the status of the aborted transaction comes out of microservice A, so a consumer would need to listen to both the output of C and the failed transaction stream from A to get a complete picture of finalized transactions.

警告

记住单作者原则。不应将超过一项服务发布到事件流。

Remember the single-writer principle. No more than one service should publish to an event stream.

即使从输出和失败的交易流中进行消费，消费者仍然无法获取正在进行的交易或陷入处理过程中的交易的状态。这需要具体化每个事件流，或者通过 API 公开每个微服务的内部状态，如本章前面所述。对精心设计的事务的工作流进行更改需要处理与非事务性工作流相同的挑战，但会增加回滚工作流中先前微服务所做的更改的开销。

Even when consuming from both the output and failed transactions streams, the consumer will still not be able to get the status of ongoing transactions or of transactions that have gotten stuck in processing. This would require that each event stream be materialized, or that the internal state of each microservice be exposed via API, as discussed earlier in this chapter. Making changes to the workflow of a choreographed transaction requires dealing with the same challenges as a nontransactional workflow, but with the added overhead of rolling back changes made by the previous microservice in the workflow.

精心设计的交易可能有些脆弱，通常需要严格的排序，并且在监控方面可能存在问题。它们最适合具有极少数微服务的服务，例如具有非常严格的排序且需要工作流更改的可能性较低的一对或三个微服务。

Choreographed transactions can be somewhat brittle, generally require a strict ordering, and can be problematic to monitor. They work best in services with a very small number of microservices, such as a pair or a trio with very strict ordering and a low likelihood of needing workflow changes.

精心安排的交易

Orchestrated Transactions

精心安排的事务建立在协调器模型的基础上，并添加了从工作流中的任何点恢复事务的逻辑。您可以通过反转工作流逻辑并确保每个工作微服务可以提供补充的反转操作来回滚这些事务。

Orchestrated transactions build on the orchestrator model, with the addition of logic to revert the transaction from any point in the workflow. You can roll back these transactions by reversing the workflow logic and ensuring that each worker microservice can provide a complementary reversing action.

精心安排的事务还可以支持各种信号，例如超时和人工输入。您可以使用超时来定期检查本地物化状态，以了解事务处理的时间。通过 REST API（参见第 13 章）的人工输入可以与其他事件一起处理，根据需要处理取消指令。协调器的集中式性质允许密切监控任何给定事务的进度和状态。

Orchestrated transactions can also support a variety of signals, such as timeouts and human inputs. You can use timeouts to periodically check the local materialized state to see how long a transaction has been processing. Human inputs via a REST API (see Chapter 13) can be processed alongside other events, handling cancellation instructions as required. The centralized nature of the orchestrator allows for close monitoring of the progress and state of any given transaction.

由于工作微服务之一的返回值、超时或人工操作员发送的中断，事务可能在工作流中的任何点中止。

The transaction can be aborted at any point in the workflow due to a return value from one of the worker microservices, a timeout, or an interrupt sent from a human operator.

一个简单的两阶段协调事务拓扑如图 8-7所示。

A simple two-stage orchestrated transaction topology is shown in Figure 8-7.

图 8-7。简单的协调事务拓扑

事件从输入流中消耗并由编排器处理。在此示例中，编排器使用对工作流微服务的直接请求-响应调用。向服务 A 发出请求，协调器在等待响应时阻塞。一旦获得响应，编排器就会更新其内部状态并调用服务B，如图8-8所示。

Events are consumed from the input stream and processed by the orchestrator. In this example, the orchestrator is using direct request-response calls to the workflow’s microservices. A request is made to service A, and the orchestrator blocks while awaiting a response. Once it obtains the response, the orchestrator updates its internal state and calls service B, as shown in Figure 8-8.

图 8-8。简单的协调事务，但事务失败

服务 B 无法执行必要的操作，并且在用尽自己的重试和错误处理逻辑后，最终向协调器返回失败响应。编排器现在必须根据该事件的当前状态制定回滚逻辑，确保它向所有所需的微服务发出回滚命令。

Service B cannot perform the necessary operation and, after exhausting its own retries and error-handling logic, eventually returns a failure response to the orchestrator. The orchestrator must now enact its rollback logic based on the current state of that event, ensuring that it issues rollback commands to all required microservices.

图 8-9演示了协调器向服务 A 发出回滚命令（服务 B 的失败响应表明它没有向其内部数据存储写入任何内容）。在此示例中，服务 A 成功执行回滚，但如果在回滚期间失败，则将由编排器确定下一步要做什么。编排器可以多次重新发出命令，通过监控框架发出警报，或者终止应用程序以防止出现进一步的问题。

Figure 8-9 demonstrates the orchestrator issuing a rollback command to service A (service B’s failure response indicates it did not write anything to its internal data store). In this example, Service A performs the rollback successfully, but if it were to fail during its rollback, it would be up to the orchestrator to determine what to do next. The orchestrator could reissue the command a number of times, issue alerts via monitoring frameworks, or terminate the application to prevent further issues.

图 8-9。在协调的事务中发出回滚命令

事务回滚后，协调器将决定下一步做什么以完成该事件的处理。它可能会多次重试该事件、丢弃该事件、终止应用程序或输出失败事件。协调器作为单一生产者，将事务失败发布到输出流，并让下游消费者处理它。这与编排不同，在编排中，没有单个流可以在不放弃单写入器原则的情况下消耗所有输出。

Once the transaction has been rolled back, it is up to the orchestrator to decide what to do next to finalize the processing of that event. It may retry the event a number of times, discard it, terminate the application, or output a failure event. The orchestrator, being the single producer, publishes the transaction failure to the output stream and lets a downstream consumer handle it. This differs from choreography, where there is no single stream from which to consume all output without discarding the single writer principle.

编排器还可以在同一输出流中公开正在进行的事务的状态，在工作服务返回结果时更新事务实体。这可以提供对底层事务状态的高度可见性，并允许基于流的监控。

The orchestrator can also expose the status of transactions in progress in the same output stream, updating the transaction entity as worker services return results. This can provide high visibility into the state of underlying transactions and allows for stream-based monitoring.

与精心设计的事务相比，精心安排的事务可以更好地了解工作流依赖性、更大的更改灵活性以及更清晰的监控选项。编排器实例增加了工作流的开销并需要 管理，但可以为复杂的工作流提供精心设计的事务无法提供的清晰度和结构。

Orchestrated transactions offer better visibility into workflow dependencies, more flexibility for changes, and clearer monitoring options than choreographed transactions. The orchestrator instance adds overhead to the workflow and requires management, but can provide to complex workflows the clarity and structure that choreographed transactions cannot provide.

确保有界上下文的严格成员资格

Ensure Strict Membership to a Bounded Context

组成解决方案的函数和内部事件流必须严格属于有界上下文，以便清楚地识别函数和数据的所有者。在大量实施微服务解决方案时，组织通常会遇到有关功能、服务和事件流的所有权问题 。虽然许多微服务解决方案以 1:1 的方式映射到有界上下文，但n :1 映射并不罕见，因为单个有界上下文可以使用多个函数。确定哪个函数属于哪个有界上下文非常重要，因为函数的高粒度可能会模糊这些界限。

The functions and internal event streams composing a solution must strictly belong to a bounded context, such that the owner of the function and data is clearly identified. It is common for an organization to have ownership questions around functions, services, and event streams when implementing microservice solutions in large numbers. While many microservice solutions map 1:1 to a bounded context, an n:1 mapping is not uncommon, as multiple functions may be used for a single bounded context. It’s important to identify which function belongs to which bounded context, because the high granularity of functions can blur those lines.

使用函数维护有界上下文的一些实用方法包括：

Some practical ways of maintaining bounded contexts with functions include:

• 确保数据存储对外部上下文保密。

• Ensure that data stores are kept private from external contexts.

• 耦合到其他上下文时，使用标准请求响应或事件驱动接口。

• Use standard request-response or event-driven interfaces when coupling to other contexts.

• 围绕哪个功能属于哪个上下文维护严格的元数据（功能到产品的 1:1 映射）。

• Maintain strict metadata around which function belongs to which context (a 1:1 mapping of function to product).

• 在映射到有界上下文的存储库中维护函数代码。

• Maintain function code within a repository mapped to the bounded context.

仅在处理完成后提交偏移量

Commit Offsets Only After Processing Has Completed

偏移量在以下两个时间之一提交：函数启动时或函数完成处理时。仅在给定事件或一批事件的处理完成后提交偏移量是 FaaS 最佳实践。了解如何处理基于函数的特定解决方案的偏移对您来说非常重要，因此让我们看一下每种方法的含义。

Offsets are committed at one of two times: when the function starts or when the function has completed its processing. Committing offsets only after processing has completed for a given event or batch of events is a FaaS best practice. It is important for you to understand how offsets for your specific function-based solutions are handled, so let’s take a look at the implications of each approach.

少即是多

Less Is More

FaaS 框架的一个经常被引用的功能是它们可以轻松编写单个函数并在多个服务中重用它。然而，采用这种方法可能会导致解决方案高度分散，从而难以准确辨别有界上下文中发生的情况。此外，给定功能的所有权变得不明确，并且可能不清楚功能的更改是否会对其他服务产生负面影响。虽然功能的版本控制可以帮助解决这个问题，但当多个产品需要维护和改进功能的不同版本时，它也可能导致冲突。

An often-cited feature of FaaS frameworks is that they make it easy to write a single function and reuse it in multiple services. Following this approach, however, can lead to a highly fragmented solution that makes it difficult to discern exactly what is going on within the bounded context. Additionally, ownership of a given function becomes ambiguous, and it may not be clear if changes to a function could negatively affect other services. While versioning of functions can help with this issue, it can also lead to conflicts when multiple products need to maintain and improve different versions of a function.

FaaS 解决方案可能会合并多个函数来解决有界上下文的业务需求，虽然这并不是一种罕见或不好的做法，但 FaaS 解决方案的一个好的经验法则是，更少的函数比许多细粒度的函数更好。测试、调试和管理一个功能比对多个功能进行相同的操作要容易得多。

FaaS solutions may incorporate multiple functions to solve the business requirements of the bounded context, and while this is not an uncommon or bad practice, a good rule of thumb for FaaS solutions is that fewer functions are better than many granular functions. Testing, debugging, and managing just one function is much easier than doing the same for multiple functions.

基于新事件的触发：事件流监听器

Triggering Based on New Events: The Event-Stream Listener

当事件产生到事件流中时可以触发函数。事件流侦听器触发器将事件消耗隔离在预定义使用者后面，从而减少开发人员必须编写的开销代码量。事件以事件数组的形式直接注入到函数中，按照事件流的顺序排列，或者如果从队列中消费，则作为无序事件的集群。您可以创建从事件流到函数的多个映射，以便函数可以使用来自许多不同流的事件。

Functions can be triggered when an event is produced into an event stream. The event-stream listener trigger isolates event consumption behind a predefined consumer, reducing the amount of overhead code that a developer must write. Events are injected directly into the function in a form of an array of events, in sequential order from an event stream, or as a cluster of unordered events if consuming from a queue. You can create multiple mappings from event streams to functions, such that a function can consume events from many different streams.

Google、Microsoft 和 Amazon 的 FaaS 解决方案提供了此触发器，供其专有事件代理使用，但目前不支持直接从开源代理触发。该方法的一般结构如图 9-1所示。

FaaS solutions from Google, Microsoft, and Amazon provide this trigger for usage with their proprietary event brokers, but currently do not support triggering directly from open source brokers. The generalized structure of this approach is shown in Figure 9-1.

图 9-1。与 FaaS 框架集成的事件流监听器

相反，OpenFaaS、Kubeless、Nuclio 等开源解决方案提供了各种带有各种事件代理的触发插件，例如 Kafka、Pulsar 和 NATS 等。例如，Apache Kafka Connect允许您触发第三方FaaS框架的功能。由于 Kafka Connect 在 FaaS 框架之外运行，因此您最终会得到如图 9-2所示的事件流侦听器。

Conversely, open source solutions such as OpenFaaS, Kubeless, Nuclio, and others provide a variety of triggering plug-ins with various event brokers, such as Kafka, Pulsar, and NATS, to name a few. For instance, Apache Kafka Connect allows you to trigger the functions of third-party FaaS frameworks. Since Kafka Connect runs outside of the FaaS framework, you would end up with an event-stream listener as per Figure 9-2.

图 9-2。Kafka Connect 提供的外部事件流侦听器应用程序

尽管前面的示例中未显示，但函数结果可以输出到它们自己的事件流，不仅用于输出数据，还用于跟踪函数的成功。

Though not shown in the previous examples, function results can be output to their own event streams, not just for the purpose of outputting data but also for tracking the function’s success.

同步触发器要求函数在发出下一个事件之前完成。这对于维护处理顺序尤其重要，并且受到正在处理的事件流的并行性的限制。相反，异步触发可以向多个函数发出多个事件，每个函数在完成时都会报告。但是，这不会维护处理顺序，并且仅当处理顺序对业务逻辑不重要时才应使用。

Synchronous triggers require the function to complete before they issue the next events. This is particularly important for maintaining the processing order and is limited by the parallelism of the event stream being processed. Conversely, asynchronous triggering can issue multiple events to multiple functions, each one reporting back as it is completed. This will not maintain the processing order, however, and should be used only when processing order is not important to the business logic.

批处理大小和批处理窗口是流侦听器触发器中需要考虑的两个重要属性。批处理大小指示要分派处理的最大事件数，而批处理窗口指示等待其他事件（而不是立即触发函数）的最长时间。这两个参数都用于确保启动函数的开销分散在批量记录中，以降低成本。

Batch size and batch window are two important properties to consider in stream-listener triggers. The batch size dictates the maximum number of events to dispatch for processing, while the batch window indicates the maximum amount of time to wait for additional events, instead of triggering the function immediately. Both of these parameters are used to ensure that the overhead of starting the function is spread among the batch of records to reduce costs.

由流侦听器触发器执行的函数如下所示：

A function executed by a stream-listener trigger looks something like the following:

  public int myEventfunction(Event[] events, Context context) {
    for(Event event: events)
        try {
          println (event.key + ", " + event.value);
        } catch (Exception e) {
          println ("error printing " + event.toString);
        }
    //Indicates to the FaaS framework that batch processing was completed.
    context.success();
    return 0;
  }

  public int myEventfunction(Event[] events, Context context) {
    for(Event event: events)
        try {
          println (event.key + ", " + event.value);
        } catch (Exception e) {
          println ("error printing " + event.toString);
        }
    //Indicates to the FaaS framework that batch processing was completed.
    context.success();
    return 0;
  }

基于消费者组滞后的触发

Triggering Based on Consumer Group Lag

消费者组的滞后指标是触发函数的另一种方式。您可以通过定期轮询单个应用程序的消费者组的偏移量并计算当前消费者偏移量与流的头部偏移量之间的增量来检测延迟（有关延迟监控的更多信息，请参阅“消费者偏移延迟监控” ）。与流侦听器触发器类似，滞后监控也可用于扩展非 FaaS 微服务。

A consumer group’s lag metric is another way to trigger functions. You can detect lag by periodically polling the offsets of an individual application’s consumer groups and computing the delta between the current consumer offset and the head offset of the stream (see “Consumer Offset Lag Monitoring” for more on lag monitoring). While similar to the stream listener trigger, lag monitoring can also be used for scaling non-FaaS microservices.

滞后监控通常涉及计算滞后指标并将其报告给您选择的监控框架。然后，监控框架可以调用FaaS框架来告诉它启动在事件流上注册的功能。高滞后值可以指示可以启动多个函数实例以更快地处理负载，而低滞后值可能只需要单个函数实例来处理积压。您可以逐个微服务定制滞后量和功能启动之间的关系，确保符合 SLA。

Lag monitoring typically involves computing and reporting lag metrics to your monitoring framework of choice. The monitoring framework can then call the FaaS framework to tell it to start up the functions registered on the event stream. A high lag value can indicate that multiple function instances can be started to more quickly process the load, while a low lag value may require only a single function instance to process the backlog. You can tailor the relationship between lag quantity and function startup on a microservice-by-microservice basis, ensuring compliance with SLAs.

前面提到的事件流侦听器触发器与这一事件流侦听器触发器之间的主要区别之一是，通过滞后触发，该函数直到启动后才会消耗事件。由滞后触发器启动的函数具有更广泛的职责范围，包括与事件代理建立客户端连接、使用事件以及提交回任何偏移更新。这使得由滞后触发的函数与基本的生产者/消费者客户端更加相似，尽管其生命周期有限。以下示例函数说明了此 工作流程：

One of the major differences between the previously mentioned event-stream listener trigger and this one is that with lag triggering, the function does not consume the events until after it is started. Functions started by the lag trigger have a wider domain of responsibilities, including establishing a client connection with the event broker, consuming the events, and committing back any offset updates. This makes functions triggered by lag much more similar to basic producer/consumer clients, albeit with a limited lifespan. The following example function illustrates this workflow:

  public int myLagConsumerfunction(Context context) {
    String consumerGroup = context.consumerGroup;
    String streamName = context.streamName;

    EventBrokerClient client = new EventBrokerClient(consumerGroup, ...);

    Event[] events = client.consumeBatch(streamName, ...);

    for(Event event: events) {
        // Do event processing work
        doWork(event);
    }
    //Commit the offsets back to the event broker
    client.commitOffsets();

    //Indicates to the FaaS framework that the function succeeded.
    context.success();

    //Return, letting the lag-triggering system know it was a success
    return 0;
  }

  public int myLagConsumerfunction(Context context) {
    String consumerGroup = context.consumerGroup;
    String streamName = context.streamName;

    EventBrokerClient client = new EventBrokerClient(consumerGroup, ...);

    Event[] events = client.consumeBatch(streamName, ...);

    for(Event event: events) {
        // Do event processing work
        doWork(event);
    }
    //Commit the offsets back to the event broker
    client.commitOffsets();

    //Indicates to the FaaS framework that the function succeeded.
    context.success();

    //Return, letting the lag-triggering system know it was a success
    return 0;
  }

消费者组和流名称作为上下文中的参数传递。创建客户端，使用和处理事件，并将偏移量提交回事件代理。该函数将success结果指示回FaaS框架，然后返回。

The consumer group and stream name are passed in as parameters in the context. The client is created, events are consumed and processed, and the offsets are committed back to the event broker. The function indicates a success result back to the FaaS framework and then returns.

如果该函数经常由滞后监视器触发，则很有可能它在上次迭代中仍然是热的，并且连接到事件代理客户端的开销可能不是一个因素。当然，这取决于客户端和事件代理配置使用的超时。对于较长时间的不活动，消费者组重新平衡和客户端冷启动将稍微减少函数实例可以处理的工作量。

If the function is frequently triggered by the lag monitor, there is a good chance that it will still be warm from the last iteration, and the overhead of connecting to the event broker client may not be a factor. This, of course, depends on the timeouts used by the client and event broker configurations. For longer periods of inactivity, consumer group rebalancing and client cold starts will slightly reduce the amount of work that a function instance can process.

按计划触发

Triggering on a Schedule

还可以安排功能定期在特定日期时间启动。计划的函数以指定的时间间隔启动，轮询源事件流以查找新事件，并根据需要进行处理或关闭。轮询周期应保持较短，以便维持 SLA，但轮询过于频繁可能会给 FaaS 框架和事件代理带来过度的负载。

Functions can also be scheduled to start up periodically and at specific datetimes. The scheduled functions start up at the specified interval, poll the source event-streams for new events, and process them or shut down as necessary. The polling period should be kept low so that SLAs are maintained, but polling too frequently may put undue load on both the FaaS framework and event broker.

基于时间的触发器的客户端代码看起来与消费者组滞后触发器示例的客户端代码相同。

Client code for a time-based trigger looks identical to that of the consumer group lag trigger example.

使用 Webhooks 触发

Triggering Using Webhooks

函数还可以通过直接调用来触发，从而允许与监控框架、调度程序和其他第三方应用程序进行自定义集成。

Functions can also be triggered by direct invocation, allowing custom integration with monitoring frameworks, schedulers, and other third-party applications.

触发资源事件

Triggering on Resource Events

对资源所做的更改也可能是触发因素的来源。例如，创建、更新或删除文件系统中的文件可以触发函数，就像对数据存储中的行进行相同的修改一样。由于事件驱动的微服务域中的大多数事件都是通过事件流生成的，因此这种特定的资源触发器在大多数业务工作流中并不经常使用。然而，当您与需要 FTP 或其他文件服务将文件放入其中的外部数据源集成时，它非常有用。

Changes made to resources can also be a source of triggers. For instance, creating, updating, or deleting a file in a filesystem can trigger functions, as can the same modifications made to a row in a data store. Since most of the events in the event-driven microservice domain are generated via event streams, this particular resource trigger isn’t often used in most business workflows. It is, however, quite useful when you are integrating with external sources of data that require an FTP or other file service to drop their files into.

事件驱动的通信模式

Event-Driven Communication Pattern

一个函数的输出可以生成到另一个消费函数的事件流中。有界上下文可以由许多函数和许多内部事件流组成，每个函数定义具有不同的触发和缩放逻辑。每个函数都以自己的速率处理传入事件，并相应地消耗事件、执行工作并产生输出。这种设计的一个例子如图 9-3所示。

The output of one function can be produced into an event stream for another consuming function. A bounded context may be made up of many functions and many internal event streams, with varying triggering and scaling logic for each function definition. Each function processes incoming events at its own rate, with events being consumed, work being performed, and outputs produced accordingly. An example of this design is shown in Figure 9-3.

图 9-3。代表单个微服务的多功能事件驱动 FaaS 拓扑

在此示例中，功能 A 的触发独立于功能 B 和 C 的触发器。事件流 2 和 3 被视为内部事件流，对其内容的访问完全仅限于有界上下文之外的任何函数。每个函数都使用相同的使用者组来使用来自其源流的事件，因为这些函数都位于同一有界上下文中。这确保了这些功能以与非基于 FaaS 的微服务相同的方式有效运行。

In this example, function A is triggered independently of the triggers for functions B and C. Event streams 2 and 3 are considered internal event streams, with access to their contents completely restricted to any functions outside of the bounded context. Each function consumes events from its source stream using the same consumer group, as the functions are all colocated within the same bounded context. This ensures that the functions are effectively operating in the same way as a non-FaaS-based microservice.

使用基于事件的通信模式有几个好处。拓扑中的每个函数都可以管理自己的消费者偏移量，一旦其工作完成就提交每个偏移量。在事件流处理之外，功能之间不需要进行协调。上图显示了基于编排的设计模式，但也可以使用编排。此外，事件处理中的任何失败都不会导致任何数据丢失，因为事件会持久存储在事件代理中，并且只会由下一个函数实例重新处理。

There are several benefits of using an event-based communication pattern. Each function within the topology can manage its own consumer offsets, committing each offset once its work is done. No coordination between functions needs to occur outside of the event stream processing. The preceding figure shows a choreography-based design pattern, though orchestration can also be used. Additionally, any failures in the event processing will not result in any data loss, as the events are durably stored in the event broker and will simply be reprocessed by the next function instance.

直接呼叫模式

Direct-Call Pattern

在直接调用模式中，函数可以直接从其自己的代码中调用其他函数。其他函数的直接调用可以异步执行，这本质上是一种“即发即忘”的方法，也可以同步执行，其中调用函数等待返回值。

In the direct-call pattern, a function can call other functions directly from its own code. Direct invocation of other functions can be performed asynchronously, which is essentially a “fire-and-forget” approach, or synchronously, where the calling function awaits for a return value.

编排和异步函数调用

Choreography and asychronous function calling

异步直接调用导致了基于编排的 FaaS 解决方案。一个函数只是根据其业务逻辑调用下一个函数，然后由该函数和 FaaS 框架来处理后续步骤，包括任何失败或错误。异步直接调用函数拓扑是将函数调用链接在一起的简单方法。图 9-4给出了一个示例。

Asychronous direct calls lead to a choreography-based FaaS solution. One function simply invokes the next one based on its business logic, leaving it up to that function and the FaaS framework to handle the next steps, including any failures or errors. An asynchronous direct-call function topology is a simple way to chain function calls together. Figure 9-4 illustrates an example.

图 9-4。在有界上下文中编排异步函数调用

函数 A 在处理一批事件时调用函数 B，完成后，函数 A 可以简单地更新其使用者偏移量并终止。同时，函数 B 继续其处理并向输出事件流产生任何输出。

Function A invokes function B as it processes its batch of events, and once done, function A can simply update its consumer offsets and terminate. Meanwhile, function B continues its processing and produces any outputs to the output event stream.

异步直接调用的一个主要缺点是确保仅在成功处理的情况下更新消费者偏移量。在示例中，函数 B 没有对函数 A 的反馈，因此只有函数 A 中的错误才能防止工作流错误地提交消费者组偏移量。但是，丢失事件对于某些工作流程可能并不重要，在这些情况下，可以 忽略此问题。

One major downside to asynchronous direct calls is in ensuring that the consumer offsets are updated only in the case of successful processing. In the example, function B has no feedback to function A, and so only errors in function A will prevent the workflow from incorrectly committing the consumer group offsets. However, losing events may not matter to some workflows, and in those cases this problem can be dismissed.

另一个潜在的主要问题是，由于多次调用函数 B，事件可能会被无序处理。考虑函数 A 的代码：

Another potentially major issue is that events may be processed out of order due to multiple invocations of function B. Consider the code for function A:

  public int functionA(Event[] events, Context context) {
    for(Event event: events) {
      //Do function A's processing work

      //Invoke function B asynchronously per event
      //Does not wait for return value
      asyncfunctionB(event);
    }
    context.success();
    return 0;
  }

  public int functionA(Event[] events, Context context) {
    for(Event event: events) {
      //Do function A's processing work

      //Invoke function B asynchronously per event
      //Does not wait for return value
      asyncfunctionB(event);
    }
    context.success();
    return 0;
  }

函数 B 是与函数 A 的工作内联调用的。根据您的 FaaS 框架，这可能会导致创建函数 B 的多个实例，每个实例都独立于其他实例运行。这将产生竞争条件，其中函数 B 的某些执行将先于其他执行完成，从而可能导致事件处理无序。

Function B is called inline with the work from function A. Depending on your FaaS framework, this may result in multiple instances of function B being created, each of which runs independently of the others. This will create a race condition where some executions of function B will finish before others, potentially leading to out-of-order processing of events.

同样，按如下方式编写代码也无法解决排序问题。处理仍然会无序发生，因为函数 A 的处理工作将在函数 B 执行之前对批处理中的所有事件执行。

Similarly, writing your code as follows will not solve ordering problems either. Processing will still happen out of order, as function A’s processing work will be executed for all events in the batch prior to function B’s execution.

  public int functionA(Event[] events, Context context) {
    for(Event event: events) {
      //Do function A's processing work
    }
    //Invoke function B asynchronously with entire batch of events
    asyncFunctionB(events);

    context.success()
    return 0;
  }

  public int functionA(Event[] events, Context context) {
    for(Event event: events) {
      //Do function A's processing work
    }
    //Invoke function B asynchronously with entire batch of events
    asyncFunctionB(events);

    context.success()
    return 0;
  }

对于每个事件，按顺序处理要求在处理下一个事件之前严格执行函数 A，然后执行 B。一个事件必须处理完毕才能开始下一个事件；否则，可能会导致不确定的行为。如果函数 A 和 B 作用于同一外部数据存储，则尤其如此，因为函数 B 可能依赖于函数 A 写入的数据。

In-order processing requires strictly executing function A before B, for each event, before processing the next event. An event must be completely processed before the next one can be started; otherwise, nondeterministic behavior will likely result. This is particularly true if functions A and B are acting on the same external data store, as function B may rely on data written by function A.

在许多情况下，异步调用不足以满足有界上下文的需求。在这些情况下，请考虑以编排方式使用同步调用是否更合适。

In many cases, asynchronous calls are not sufficient for the needs of the bounded context. In these cases, consider whether using synchronous calls in an orchestrated manner is more suitable.

编排和同步函数调用

Orchestration and synchronous function calling

同步函数调用允许您调用其他函数并等待结果，然后再继续其余的业务逻辑。这允许实现编排模式，如第 8 章所述。

Synchronous function calls allow you to invoke other functions and await the results before proceeding with the remaining business logic. This allows for the implementation of the orchestration pattern, as covered in Chapter 8.

事件流——触发处理

Event stream—triggered processing

在以下示例中，当新事件到达分区事件流时，将触发单个编排函数。该函数启动并开始处理输入的事件批次，为每个函数顺序调度事件。图 9-5显示了单个有界上下文中基于函数的编排示例。

In the following example, a single orchestration function is triggered when new events arrive in a partitioned event stream. The function starts up and begins processing the input batch of events, dispatching event sequentially for each function. Figure 9-5 shows an example of function-based orchestration within a single bounded context.

图 9-5。在有界上下文中协调同步函数调用

以下是编排代码的示例：

Here is an example of the orchestration code:

  public int orchestrationFunction(Event[] events, Context context) {
    for(Event event: events) {
      //Synchronous function Calls
      Result resultFromA = invokeFunctionA(event);
      Result resultFromB = invokeFunctionB(event, resultFromA);
      Output output = composeOutputEvent(resultFromA, resultFromB);
      //Write to the output stream
      producer.produce("Output Stream", output);
    }
    //This will tell the FaaS framework to update the consumer offsets
    context.success();
    return 0;
  }

  public int orchestrationFunction(Event[] events, Context context) {
    for(Event event: events) {
      //Synchronous function Calls
      Result resultFromA = invokeFunctionA(event);
      Result resultFromB = invokeFunctionB(event, resultFromA);
      Output output = composeOutputEvent(resultFromA, resultFromB);
      //Write to the output stream
      producer.produce("Output Stream", output);
    }
    //This will tell the FaaS framework to update the consumer offsets
    context.success();
    return 0;
  }

编排函数按顺序调用函数 A 和 B，并等待每个函数的结果。如果需要，可以将函数 A 的输出发送到 B。每个事件在下一个事件开始之前都通过工作流程进行完全处理，确保遵守偏移顺序。一旦消费者函数完成处理批次中的每个事件，它就可以发出成功消息并相应地更新偏移量。

The orchestration function invokes functions A and B in order and awaits the results from each function. The output of function A can be sent to B if needed. Each event is fully processed through the workflow before the next one is started, ensuring that offset order is respected. Once the consumer function has completed processing each event in the batch, it can issue a success message and update the offsets accordingly.

队列触发的事件处理

Queue-triggered event processing

如果您使用具有单独提交功能的队列，则触发机制可以简单地为每个事件触发单独的编排功能。协调器完成工作后需要将处理确认提交回队列。如果编排器无法处理工作，它将由下一个创建的编排器实例来接替。

If you use a queue with individual commit capabilities, the triggering mechanism can simply trigger an individual orchestration function for each event. The orchestrator will need to commit the processing confirmation back to the queue after it has completed its work. In the case that the orchestrator fails to process the work, it will simply be picked up by the next orchestrator instance created.

分配足够的资源

Allocating Sufficient Resources

每个函数都可以分配特定数量的 CPU 和内存。根据您的功能需求调整这些参数非常重要；过度分配的成本可能很高，而分配不足可能会导致函数崩溃或需要很长时间才能完成。

Each function can be allocated a specific amount of CPU and memory. It is important to tune these parameters to the needs of your function; overallocation can be expensive, while underallocation may result in your functions crashing or taking too long to complete.

最大执行时间是另一个因素，因为它限制了函数可以运行的时间。该参数与批量大小密切相关，因为函数处理事件所需的时间通常与要处理的事件数量线性相关。将最大执行时间设置为高于处理特定批量大小的事件的最大预期时间，以避免不必要的函数超时错误。

Maximum execution time is another factor, as it limits how long a function may run. This parameter is closely related to the batch size, as the time a function needs to process events is very often linearly related, on average, with the number of events to process. Set the maximum execution time higher than the maximum expected time to process a specific batch size of events to avoid unnecessary function timeout errors.

最后，您必须考虑属于基于函数的解决方案的有界上下文中的状态存储的任何外部 I/O。函数的工作负载随输入事件流的变化而变化，某些工作负载需要与外部状态一致的 I/O，而其他工作负载仅需要零星的 I/O。无法提供足够的 I/O 资源可能会导致吞吐量和性能下降。

Lastly, you must consider any external I/O to state stores belonging within the bounded context of the function-based solution. The workload of a function varies with the flow of input events, with some workloads requiring consistent I/O to external state, and other workloads requiring only sporadic I/O. A failure to provide sufficient I/O resources can result in degraded throughput and performance.

批量事件处理参数

Batch Event-Processing Parameters

如果函数在其执行生命周期内无法处理为其分配的一批事件，则该函数的执行将被视为失败，并且必须重新处理该批事件。但是，除非对函数分配的执行时间或输入事件的批量大小进行任何更改，否则它很可能会再次失败。因此，必须发生以下两种情况之一：

If a function is unable to process its assigned batch of events during its execution lifespan, the execution of the function is considered to have failed and the batch must be processed again. However, barring any changes to the function’s allocated execution time or the batch size of input events, it is likely to simply fail again. Therefore, one of two things must occur:

• 增加函数的最大执行时间。

• Increase the maximum execution time of the function.

• 减小函数处理的事件的最大批量大小。

• Decrease the maximum batch size of events processed by the function.

此外，某些事件侦听器触发系统（例如 Amazon 和 Microsoft 提供的系统）允许您选择在失败时自动将批处理大小减半并重新执行失败的函数。随后的错误会导致输入批次再次减半并重新执行函数，直到达到可以按时完成处理的程度。

Additionally, some event-listener triggering systems, such as those provided by Amazon and Microsoft, give you the option to automatically halve the batch size on failure and re-execute the failed function. Subsequent errors result in the input batch being halved again and the function re-executed, until it reaches the point where it can complete its processing on time.